package service

import (
	"encoding/json"
	"errors"
	"fmt"
	"github.com/gofrs/uuid"
	"gorm.io/gorm"
	"os"
	"reflect"
	"strconv"
	"strings"
	"sync"
	"time"
	"vrcm/src/vrcm/app/admin/models"
	liveDto "vrcm/src/vrcm/app/admin/service/dto/live"
	mediaProcService "vrcm/src/vrcm/app/admin/service/media_proc"
	ext "vrcm/src/vrcm/envconfig"

	"vrcm/src/vrcm/app/admin/models/live_schedule"
	baiduService "vrcm/src/vrcm/app/admin/service/baidu"
	"vrcm/src/vrcm/app/admin/service/dto/live_schedule"
	StreamHubDto "vrcm/src/vrcm/app/admin/service/dto/stream_hub"
	liveService "vrcm/src/vrcm/app/admin/service/live"
	streamHubService "vrcm/src/vrcm/app/admin/service/stream_hub"
	"vrcm/src/vrcm/common/actions"
	"vrcm/src/vrcm/common/apis"
	cDto "vrcm/src/vrcm/common/dto"
	commonDto "vrcm/src/vrcm/common/dto"
	"vrcm/src/vrcm/common/service"
)

type ChannelTaskService struct {
	service.Service
}

// GetPage 获取ChannelTask列表
func (e *ChannelTaskService) GetPage(c *dto.ChannelTaskGetPageReq, p *actions.DataPermission, list *[]live_schedule.ChannelTask, count *int64) error {
	var err error
	var data live_schedule.ChannelTask

	err = e.Orm.Model(&data).
		Scopes(
			cDto.MakeCondition(c.GetNeedSearch()),
			cDto.Paginate(c.GetPageSize(), c.GetPageIndex()),
			actions.Permission(data.TableName(), p),
		).
		Find(list).Limit(-1).Offset(-1).
		Count(count).Error
	if err != nil {
		e.Log.Errorf("ChannelTaskService GetPage error:%s \r\n", err)
		return err
	}
	return nil
}

// Get 获取ChannelTask对象
func (e *ChannelTaskService) Get(d *dto.ChannelTaskGetReq) error {
	var data live_schedule.ChannelTask

	err := e.Orm.Model(&data).
		First(&data, d.GetId()).Error
	if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
		e.Log.Errorf("Service GetChannelTask error:%s \r\n", err)
		return err
	}

	return nil
}

// Insert 创建ChannelTask对象
func (e *ChannelTaskService) Insert(c *dto.ChannelTaskInsertReq) error {
	var err error
	var data live_schedule.ChannelTask
	c.Generate(&data)
	err = e.Orm.Create(&data).Error
	if err != nil {
		e.Log.Errorf("ChannelTaskService Insert error:%s \r\n", err)
		return err
	}
	return nil
}

// Update 修改ChannelTask对象
func (e *ChannelTaskService) Update(c *dto.ChannelTaskUpdateReq, p *actions.DataPermission) error {
	var err error
	var data = live_schedule.ChannelTask{}
	e.Orm.Scopes(
		actions.Permission(data.TableName(), p),
	).First(&data, c.GetId())
	c.Generate(&data)

	db := e.Orm.Save(&data)
	if err = db.Error; err != nil {
		e.Log.Errorf("ChannelTaskService Save error:%s \r\n", err)
		return err
	}
	if db.RowsAffected == 0 {
		return errors.New("无权更新该数据")
	}
	return nil
}

// Remove 删除ChannelTask
func (e *ChannelTaskService) Remove(d *dto.ChannelTaskDeleteReq, p *actions.DataPermission) error {
	var data live_schedule.ChannelTask

	db := e.Orm.Model(&data).
		Scopes(
			actions.Permission(data.TableName(), p),
		).Delete(&data, d.GetId())
	if err := db.Error; err != nil {
		e.Log.Errorf("Service RemoveChannelTask error:%s \r\n", err)
		return err
	}
	if db.RowsAffected == 0 {
		return errors.New("无权删除该数据")
	}
	return nil
}

func (e *ChannelTaskService) SaveTrafficOn(c *dto.TrafficOnSaveReq) error {
	var err error
	var data = live_schedule.TrafficOn{}

	//获取信源
	var basic live_schedule.SourceBasic
	err = e.Orm.Model(&live_schedule.SourceBasic{}).Where("id=?", c.SourceId).Find(&basic).Error
	if err != nil {
		e.Log.Errorf("get SourceBasic error:%s \r\n", err)
		return err
	}
	if basic.Id <= 0 {
		return errors.New("信源不存在")
	}
	//获取转码记录
	t := live_schedule.SourceTemplate{}
	err = e.Orm.Model(live_schedule.SourceTemplate{}).Where("id=?", c.TransRecordId).Find(&t).Error
	if err != nil {
		e.Log.Errorf("get SourceTemplate error:%s \r\n", err)
		return err
	}
	//获取当前正在使用的转码源
	task := live_schedule.TransCodeTask{}
	err = e.Orm.Model(live_schedule.TransCodeTask{}).Where("task_record_id = ? and url_type=?", c.TransRecordId, t.ActiveItem).Find(&task).Error
	if err != nil {
		e.Log.Errorf("get TransCodeTask error:%s \r\n", err)
		return err
	}
	if c.StreamType == 2 && task.Id <= 0 {
		return errors.New("转码任务不存在")
	}
	if len(t.TransCodeSwitchTaskId) > 0 {
		c.TransUdpUrl = t.TransCodeOutputAddress
	} else {
		c.TransUdpUrl = task.OutUdpUrl
	}

	e.Orm.First(&data, c.Id)
	c.Generate(&data)
	channelBasic := live_schedule.ChannelBasic{}
	err = e.Orm.Model(live_schedule.ChannelBasic{}).Where("id = ? ", c.ChannelId).Find(&channelBasic).Error
	if err != nil {
		e.Log.Errorf("get channelBasic error:%s \r\n", err)
		return err
	}
	if channelBasic.ChannelState != 2 {
		data.TrafficOnState = 1
	}

	db := e.Orm.Save(&data)
	if err = db.Error; err != nil {
		e.Log.Errorf("ChannelTaskService SaveTrafficOn error:%s \r\n", err)
		return err
	}
	c.Id = data.Id
	c.NodeId = data.NodeId
	return nil
}

func (e *ChannelTaskService) SaveTrafficOns(c *dto.TrafficOnsSaveReq) (*dto.TrafficOnsSaveRes, error) {

	res := &dto.TrafficOnsSaveRes{}
	tx := e.Orm.Begin()
	defer func() {
		if r := recover(); r != nil {
			e.Log.Errorf("Service SaveTrafficOns error:%s \r\n", r)
			tx.Rollback()
			//return
		}
	}()

	//保存信源组
	groupModel := live_schedule.TrafficOn{}
	groupModel.Id = c.Id
	groupModel.TenantId = c.TenantId
	groupModel.ChannelId = c.ChannelId
	groupModel.TrafficOnName = c.TrafficOnName
	groupModel.GroupName = c.GroupName
	groupModel.IsGroup = 1
	groupModel.SourceId = c.SourceId
	u, _ := uuid.NewV4()
	groupModel.NodeId = u.String()
	err := tx.Model(&groupModel).Save(&groupModel).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("create TrafficOn error:%s \r\n", err)
		return nil, err
	}

	taskIds := make([]int, 0, 5)
	for _, v := range *c.TrafficOnSaveReqs {
		taskIds = append(taskIds, v.TransRecordId)
	}

	//获取转码地址
	sourceTemplates := make([]live_schedule.SourceTemplate, 0, 5)
	err = tx.Model(&live_schedule.SourceTemplate{}).Where("id in ?", taskIds).Find(&sourceTemplates).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("get SourceTemplate error:%s \r\n", err)
		return nil, err
	}
	sourceTemplateMap := make(map[int]live_schedule.SourceTemplate)
	for _, v := range sourceTemplates {
		v := v
		sourceTemplateMap[v.Id] = v

	}
	//获取当前正在使用的转码源
	transTasks := make([]live_schedule.TransCodeTask, 0, 5)
	err = tx.Model(live_schedule.TransCodeTask{}).Where("task_record_id in ?", taskIds).Find(&transTasks).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("get TransCodeTask error:%s \r\n", err)
		return nil, err
	}
	transTaskMap := make(map[int][]live_schedule.TransCodeTask)
	for _, v := range transTasks {
		v := v
		t, ok := transTaskMap[v.TaskRecordId]
		if !ok {
			t = make([]live_schedule.TransCodeTask, 0, 5)
		}
		t = append(t, v)
		transTaskMap[v.TaskRecordId] = t
	}

	models := make([]live_schedule.TrafficOn, 0, 5)
	for _, v := range *c.TrafficOnSaveReqs {
		v := v
		if v.IsGroup == 1 {
			continue
		}
		model := live_schedule.TrafficOn{}
		now := time.Now().Unix()
		//model.Id = v.Id
		model.TenantId = c.TenantId
		model.ChannelId = c.ChannelId
		model.SourceId = v.SourceId
		model.GroupId = strconv.Itoa(groupModel.Id)
		model.TrafficOnName = v.TrafficOnName
		model.TrafficOnType = v.TrafficOnType
		model.IsGroup = v.IsGroup
		model.StreamType = v.StreamType
		model.TemplateId = v.TemplateId
		model.IsDelete = 2
		s := sourceTemplateMap[v.TransRecordId]
		for _, v := range transTaskMap[v.TransRecordId] {
			if v.UrlType == s.ActiveItem {
				model.TransUdpUrl = v.OutUdpUrl
				break
			}
		}

		model.TransTaskId = v.TransTaskId
		model.TransRecordId = v.TransRecordId
		model.CreateTime = now
		model.UpdateTime = now
		u, _ := uuid.NewV4()
		model.NodeId = u.String()
		models = append(models, model)

	}
	if len(models) > 0 {
		db := tx.Save(&models)
		if err := db.Error; err != nil {
			tx.Rollback()
			e.Log.Errorf("ChannelTaskService SaveTrafficOn error:%s \r\n", err)
			return nil, err
		}
	}

	res.Id = groupModel.Id
	res.NodeId = u.String()
	ups := make([]live_schedule.TrafficOn, 0, 5)
	err = tx.Model(live_schedule.TrafficOn{}).Where("group_id=?", groupModel.Id).Find(&ups).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("get TrafficOn error:%s \r\n", err)
		return nil, err
	}

	for _, v := range ups {
		req := dto.TrafficOnSaveReq{}
		req.Id = v.Id
		req.SourceId = v.SourceId
		req.NodeId = v.NodeId
		req.TrafficOnName = v.TrafficOnName
		req.ChannelId = v.ChannelId
		req.TenantId = v.TenantId
		req.IsGroup = v.IsGroup
		s := sourceTemplateMap[v.TransRecordId]
		req.TransUdpUrl = s.TransCodeOutputAddress
		req.StreamType = v.StreamType
		req.TrafficOnType = v.TrafficOnType
		req.TransTaskId = v.TransTaskId
		req.TransRecordId = v.TransRecordId
		res.TrafficOnSaveReqs = append(res.TrafficOnSaveReqs, req)
	}

	tx.Commit()
	return res, nil
}

func (e ChannelTaskService) GetTrafficOn(c *dto.TrafficOnGetReq, count *int64) (*dto.TrafficOnGetRes, error) {

	var res dto.TrafficOnGetRes
	var data = live_schedule.TrafficOn{}
	err := e.Orm.Model(live_schedule.TrafficOn{}).Where("id=?", c.Id).Find(&data).Error
	if err != nil {
		e.Log.Errorf(" SourceGroup error:%s \r\n", err)
		return nil, err
	}

	if data.Id > 0 {
		res.TrafficOn = &data
	} else {
		return nil, nil
	}

	sourceIds := make([]int, 0, 5)
	if c.IsGroup == 1 {
		e.Orm.Model(live_schedule.SourceGroupRelation{}).Select("source_id").Where("group_id = ?", data.SourceId).Find(&sourceIds)
	} else {
		sourceIds = append(sourceIds, data.SourceId)
	}
	if len(sourceIds) <= 0 {
		return &res, nil
	}

	kind := data.Kind
	//获取信源基本详情
	basicService := SourceBasicService{}
	basicService.Orm = e.Orm
	basicService.Log = e.Log
	if c.StreamType == 1 {
		basics, err := basicService.GetPage(&dto.SourceBasicGetPageReq{Pagination: c.Pagination, SourceIds: sourceIds, IsDelete: 2, Kind: kind}, count)
		if err != nil {
			e.Log.Errorf("ChannelTaskService GetSourceBasic error:%s \r\n", err)
			return nil, err
		}
		res.SourceModelRes = basics
	} else {
		trans, err := basicService.SourcesTransCodeTaskPage(&dto.GetSourceTaskPageReq{Pagination: c.Pagination, SourceIds: sourceIds, TemplateId: data.TemplateId, TransTaskId: data.TransTaskId}, count)
		if err != nil {
			e.Log.Errorf("get SourcesTransCodeTaskPage error:%s \r\n", err)
			return nil, err
		}
		res.SourceTemplateModels = trans
	}

	return &res, nil
}

func (e *ChannelTaskService) GetTrafficOns(c *dto.TrafficOnGetReq, count *int64) (*dto.TrafficOnGetRes, error) {

	var res dto.TrafficOnGetRes
	var data = live_schedule.TrafficOn{}
	err := e.Orm.Model(live_schedule.TrafficOn{}).Where("id=?", c.Id).Find(&data).Error
	if err != nil {
		e.Log.Errorf(" SourceGroup error:%s \r\n", err)
		return nil, err
	}

	if data.Id > 0 {
		res.TrafficOn = &data
	} else {
		return nil, nil
	}

	//获取组名称
	var group live_schedule.SourceGroup
	err = e.Orm.Model(live_schedule.SourceGroup{}).Where("id=?", data.SourceId).Find(&group).Error
	if err != nil {
		e.Log.Errorf(" SourceGroup error:%s \r\n", err)
		return nil, err
	}

	//获取信源组下面的上车点
	ups := make([]live_schedule.TrafficOn, 0, 5)
	err = e.Orm.Model(live_schedule.TrafficOn{}).Where("group_id=?", c.Id).Find(&ups).Error
	if err != nil {
		e.Log.Errorf(" GetTrafficOn error:%s \r\n", err)
		return nil, err
	}

	res.TrafficOns = &ups
	sourceIds := make([]int, 0, 5)
	for _, v := range ups {
		sourceIds = append(sourceIds, v.SourceId)
	}
	if len(sourceIds) <= 0 {
		return &res, nil
	}

	//获取信源基本详情
	basicService := SourceBasicService{}
	basicService.Orm = e.Orm
	basicService.Log = e.Log
	basics, err := basicService.GetPage(&dto.SourceBasicGetPageReq{Pagination: c.Pagination, SourceIds: sourceIds, IsDelete: 2, IsShare: 2}, count)
	if err != nil {
		e.Log.Errorf("ChannelTaskService GetSourceBasic error:%s \r\n", err)
		return nil, err
	}
	res.SourceModelRes = basics

	return &res, nil
}

func (e *ChannelTaskService) RemoveTrafficOn(d *dto.TrafficOnDeleteReq) error {
	var data live_schedule.TrafficOn

	hub := streamHubService.StreamHubService{}
	hub.Log = e.Log
	sourceBasicService := SourceBasicService{}
	sourceBasicService.Log = e.Log
	sourceBasicService.Orm = e.Orm
	err := e.Orm.Model(&live_schedule.TrafficOn{}).Where("id=?", d.Id).Find(&data).Error
	if err != nil {
		e.Log.Errorf("RemoveTrafficOn error:%s \r\n", err)
		return err
	}

	//如果是自定义信源就删除
	if data.Kind == 2 {
		var basic live_schedule.SourceBasic
		err = e.Orm.Model(&live_schedule.SourceBasic{}).Where("id=?", data.SourceId).Find(&basic).Error
		if err != nil {
			e.Log.Errorf("get SourceBasic error:%s \r\n", err)
			return err
		}
		err = sourceBasicService.DeleteCustomSource(basic)
		if err != nil {
			e.Log.Errorf("DeleteCustomSource error:%s \r\n", err)
			return err
		}
	}

	db := e.Orm.Model(&data).Delete(&data, d.Id)
	if err := db.Error; err != nil {
		e.Log.Errorf("Service RemoveGetTrafficOn error:%s \r\n", err)
		return err
	}
	return nil
}

func (e *ChannelTaskService) RemoveTrafficOns(d *dto.TrafficOnDeleteReq) error {
	var data live_schedule.TrafficOn
	err := e.Orm.Model(live_schedule.TrafficOn{}).Where("id=?", d.Id).Find(&data).Error
	if err != nil {
		e.Log.Errorf("get TrafficOn error:%s \r\n", err)
		return err
	}

	//获取组下面的信源
	ups := make([]live_schedule.TrafficOn, 0, 5)
	err = e.Orm.Model(live_schedule.TrafficOn{}).Where("group_id=?", data.Id).Find(&ups).Error
	if err != nil {
		e.Log.Errorf("get TrafficOn error:%s \r\n", err)
		return err
	}

	ids := make([]int, 0, 5)
	for _, up := range ups {
		up := up
		ids = append(ids, up.Id)
	}
	ids = append(ids, data.Id)
	db := e.Orm.Model(&data).Where("id in ?", ids).Delete(&live_schedule.TrafficOn{})
	if err := db.Error; err != nil {
		e.Log.Errorf("Service RemoveGetTrafficOn error:%s \r\n", err)
		return err
	}
	return nil
}

func (e *ChannelTaskService) SaveMediaProcess(c *dto.MediaProcessSaveReq) error {
	var err error
	var data = live_schedule.MediaProcess{}
	e.Orm.First(&data, c.Id)
	c.Generate(&data)

	db := e.Orm.Save(&data)
	if err = db.Error; err != nil {
		e.Log.Errorf("ChannelTaskService SaveMediaProcess error:%s \r\n", err)
		return err
	}
	c.Id = data.Id
	c.NodeId = data.NodeId
	return nil
}

func (e *ChannelTaskService) GetMediaProcess(d *dto.MediaProcessGetReq) (*dto.MediaProcessModel, error) {
	var data live_schedule.MediaProcess
	model := dto.MediaProcessModel{}
	err := e.Orm.Model(&data).
		First(&data, d.Id).Error
	if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
		e.Log.Errorf("Service GetMediaProcess error:%s \r\n", err)
		return nil, err
	}
	if data.Id <= 0 {
		return nil, err
	}
	model.MediaProcess = data

	upIds := make([]int, 0, 5)

	err = e.Orm.Model(&live_schedule.UpMedia{}).Select("up_id").Where("media_id=?", d.Id).Find(&upIds).Error
	if err != nil {
		e.Log.Errorf("get UpMedia error:%s \r\n", err)
	}
	sourceIds := make([]int, 0, 5)
	err = e.Orm.Model(&live_schedule.TrafficOn{}).Select("source_id").Where("id in ?", upIds).Find(&sourceIds).Error
	if err != nil {
		e.Log.Errorf("get TrafficOn error:%s \r\n", err)
	}
	if data.MediaType == 3 {
		tempResults := make([]dto.AuditStatisticDetailRes, 0, 5)
		err = e.Orm.Model(&live_schedule.AuditResult{}).
			Where("tenant_id=? and source_id in ? and audit_type =1", data.TenantId, sourceIds).
			Select("source_id, violation_type,channel_id, count(*) as type_num").
			Group("source_id, violation_type,channel_id").Scan(&tempResults).Error
		if err != nil {
			e.Log.Errorf("get AuditResult error:%s \r\n", err)
			return nil, err
		}
		statisticMap := make(map[int][]dto.AuditStatisticDetailRes)
		for _, v := range tempResults {
			v := v
			s, ok := statisticMap[v.SourceId]
			if !ok {
				s = make([]dto.AuditStatisticDetailRes, 0, 5)
			}
			s = append(s, v)
			statisticMap[v.SourceId] = s
		}
		results := make([]dto.AuditStatisticRes, 0, 5)
		for k, details := range statisticMap {
			result := dto.AuditStatisticRes{}
			result.SourceId = k
			//num := 0
			t1 := 0
			t2 := 0
			t3 := 0
			t4 := 0
			t5 := 0
			t6 := 0
			for _, d := range details {
				switch d.ViolationType {
				case 1:
					t1 += d.TypeNum
				case 2:
					t2 = d.TypeNum
				case 3:
					t3 = d.TypeNum
				case 4:
					t4 = d.TypeNum
				case 5:
					t5 = d.TypeNum
				default:
					t6 = t6 + d.TypeNum
				}
			}
			result.Type1Num = t1
			result.Type2Num = t2
			result.Type3Num = t3
			result.Type4Num = t4
			result.Type5Num = t5
			result.Type6Num = t6
			results = append(results, result)
		}

		model.AuditStatisticRes = results
	}
	return &model, nil
}

func (e *ChannelTaskService) RemoveMediaProcess(d *dto.MediaProcessDeleteReq) error {
	var data live_schedule.MediaProcess

	db := e.Orm.Model(&data).Delete(&data, d.Id)
	if err := db.Error; err != nil {
		e.Log.Errorf("Service RemoveMediaProcess error:%s \r\n", err)
		return err
	}
	err := e.Orm.Model(&live_schedule.CortexOutput{}).Where("media_process_id=?", d.Id).Delete(&live_schedule.MediaProcess{}).Error
	if err != nil {
		e.Log.Errorf("delete CortexOutput error:%s \r\n", err)
		return err
	}
	return nil
}

func (e *ChannelTaskService) SaveTrafficOff(c *dto.TrafficOffSaveReq) error {
	tx := e.Orm.Begin()
	defer func() {
		if r := recover(); r != nil {
			e.Log.Errorf("Service SaveTrafficOff error:%s \r\n", r)
			tx.Rollback()
			//return
		}
	}()
	if err := tx.Error; err != nil {
		return err
	}
	var err error

	isDiff := false
	streamHubService := streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	streamHubService.Orm = e.Orm
	var data = live_schedule.TrafficOff{}
	tx.First(&data, c.Id)

	mediaInfo := live_schedule.MediaProcess{}
	err = tx.Model(&live_schedule.MediaProcess{}).Where("channel_id=?", c.ChannelId).Find(&mediaInfo).Error
	if err != nil {
		e.Log.Errorf("get MediaProcess error:%s \r\n", err)
		tx.Rollback()
		return err
	}

	if len(c.EquipCode) > 0 {
		off := live_schedule.TrafficOff{}
		err = e.Orm.Model(&live_schedule.TrafficOff{}).Where("equip_code=?", c.EquipCode).Find(&off).Error
		if err != nil {
			tx.Rollback()
			e.Log.Errorf("get TrafficOff error:%s \r\n", err)
			return err
		}
		if (off.Id > 0 && c.Id <= 0) || (off.Id > 0 && c.Id > 0 && off.Id != c.Id) {
			tx.Rollback()
			return errors.New("设备已在其他下车点使用")
		}
	}

	//比较黑白名单是否变化
	originIps := make([]string, 0, 3)
	err = tx.Model(&live_schedule.HlsBlackWhite{}).Select("ip_address").Where("off_id=?", data.Id).Find(&originIps).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("get HlsBlackWhite error:%s \r\n", err)
		return err
	}
	if data.EnableBlackWhite != c.EnableBlackWhite || data.ListType != c.ListType || len(originIps) != len(c.IPAddresses) {
		isDiff = true
	}
	for _, v := range c.IPAddresses {
		isSame := false
		for _, ip := range originIps {
			if ip == v {
				isSame = true
			}
		}
		if !isSame {
			isDiff = true
			break
		}
	}
	//获取通道基本信息
	var channelBasic live_schedule.ChannelBasic
	err = tx.Model(live_schedule.ChannelBasic{}).Where("id = ?", c.ChannelId).Find(&channelBasic).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("get ChannelBasic error:%s \r\n", err)
		return err
	}
	originIsCustom := data.IsCustom
	c.Generate(&data)
	if channelBasic.ChannelState != 2 {
		data.TrafficOffState = 1
	}
	db := tx.Save(&data)
	if err = db.Error; err != nil {
		e.Log.Errorf("ChannelTaskService SaveTrafficOff error:%s \r\n", err)
		return err
	}

	taskIds := make([]string, 0, 5)
	//处理被白名单
	err = tx.Model(&live_schedule.HlsBlackWhite{}).Where("off_id=?", data.Id).Delete(&live_schedule.HlsBlackWhite{}).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("delete HlsBlackWhite error:%s \r\n", err)
		return err
	}
	hlsBlackWhites := make([]live_schedule.HlsBlackWhite, 0, 5)
	for _, v := range c.IPAddresses {
		v := v
		hlsBlackWhite := live_schedule.HlsBlackWhite{}
		hlsBlackWhite.TenantId = c.TenantId
		hlsBlackWhite.ChannelId = c.ChannelId
		hlsBlackWhite.OffId = data.Id
		hlsBlackWhite.IpAddress = v
		hlsBlackWhites = append(hlsBlackWhites, hlsBlackWhite)
	}
	if len(hlsBlackWhites) > 0 {
		err = tx.Model(&live_schedule.HlsBlackWhite{}).Create(&hlsBlackWhites).Error
		if err != nil {
			tx.Rollback()
			e.Log.Errorf("create HlsBlackWhite error:%s \r\n", err)
			return err
		}
	}
	data.IPAddresses = hlsBlackWhites

	//存储推流地址映射
	pushInfoMap := make(map[string]dto.PushInfo)
	for _, v := range c.PushInfos {
		v := v
		key := ""
		if mediaInfo.MediaType != 2 {
			key = strconv.Itoa(v.SourceId) + models.Delimiter + strconv.Itoa(v.UpId) + models.Delimiter + strconv.Itoa(v.MediaId) + models.Delimiter + strconv.Itoa(v.OffId)
		} else {
			key = strconv.Itoa(v.CortexOutId) + models.Delimiter + strconv.Itoa(v.MediaId) + models.Delimiter + strconv.Itoa(v.OffId)
		}
		pushInfoMap[key] = v
	}

	//存储流地址映射
	pullInfoMap := make(map[string]dto.PullInfo)
	for _, v := range c.PullInfos {
		v := v
		key := ""
		if mediaInfo.MediaType != 2 {
			key = strconv.Itoa(v.SourceId) + models.Delimiter + strconv.Itoa(v.UpId) + models.Delimiter + strconv.Itoa(v.MediaId) + models.Delimiter + strconv.Itoa(v.OffId)
		} else {
			key = strconv.Itoa(v.CortexOutId) + models.Delimiter + strconv.Itoa(v.MediaId) + models.Delimiter + strconv.Itoa(v.OffId)
		}

		pullInfoMap[key] = v
	}

	channelTasks := make([]live_schedule.ChannelTask, 0, 5)
	err = tx.Model(live_schedule.ChannelTask{}).Where("channel_id=? and traffic_off_id=?", c.ChannelId, c.Id).Find(&channelTasks).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("get ChannelTask error:%s \r\n", err)
		return err
	}

	cortexTasks := make([]live_schedule.CortexTask, 0, 5)
	err = tx.Model(live_schedule.CortexTask{}).Where("channel_id=? and traffic_off_id=?", c.ChannelId, c.Id).Find(&cortexTasks).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("get CortexTask error:%s \r\n", err)
		return err
	}

	now := time.Now().Unix()
	scheduleIds := make([]string, 0, 5)
	//
	if (len(c.PushInfos) > 0 || len(c.PullInfos) > 0) && len(channelTasks) > 0 && mediaInfo.MediaType == 1 {
		for k, v := range channelTasks {
			v := v
			taskIds = append(taskIds, v.OutNodeId)
			key := strconv.Itoa(v.SourceId) + models.Delimiter + strconv.Itoa(v.TrafficOnId) + models.Delimiter + strconv.Itoa(v.MediaId) + models.Delimiter + strconv.Itoa(v.TrafficOffId)
			pushInfo, ok1 := pushInfoMap[key]
			//如果存在，并且没有转推任务直接保存
			if ok1 && len(pushInfo.PushUrl) > 0 && len(v.TaskId) <= 0 {
				*(&channelTasks[k].PushUrl) = pushInfo.PushUrl
			}

			//如果存在，并且有转推任务 需要先清除任务
			if ok1 && len(pushInfo.PushUrl) > 0 && len(v.TaskId) > 0 && v.PushUrl != pushInfo.PushUrl {
				err = streamHubService.StopStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.TaskId})
				if err != nil {
					tx.Rollback()
					e.Log.Errorf("StopStreamHub error:%s \r\n", err)
					return err
				}
				err = streamHubService.DeleteStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.TaskId})
				if err != nil {
					tx.Rollback()
					e.Log.Errorf("DeleteStreamHub error:%s \r\n", err)
					return err
				}
				scheduleIds = append(scheduleIds, v.TaskId)
				*(&channelTasks[k].TaskId) = ""
				*(&channelTasks[k].PushUrl) = pushInfo.PushUrl
			}

			pullInfo, ok2 := pullInfoMap[key]
			if ok2 && len(pullInfo.PullUrl) > 0 {
				arr := strings.Split(pullInfo.PullUrl, "/")
				if len(arr) > 1 {
					index := strings.Index(arr[1], ".")
					sub := arr[1][0:index]
					//如果编辑自定义参数且任务是启用状态需要调用编辑自定义参数接口 && channelBasic.ChannelState != 2
					if len(v.TaskId) > 0 {
						err = streamHubService.EditRefinePlayUrl(&StreamHubDto.EditRefinePlayUrlReq{TenantId: c.TenantId, Id: v.OutNodeId, RedefinePlayUrlEnable: "1", RedefinePlayUrlParams: StreamHubDto.RedefinePlayUrlParams{Path: arr[0], Sub: sub}})
						if err != nil {
							tx.Rollback()
							e.Log.Errorf("EditRefinePlayUrl error:%s \r\n", err)
							return err
						}
					}
				}
				*(&channelTasks[k].CustomOutUrl) = pullInfo.PullUrl
			}
		}

		if len(channelTasks) > 0 {
			err = tx.Save(&channelTasks).Error
			if err != nil {
				tx.Rollback()
				e.Log.Errorf("save ChannelTask error:%s \r\n", err)
				return err
			}
		}

	}

	if (len(c.PushInfos) > 0 || len(c.PullInfos) > 0) && mediaInfo.MediaType == 2 {
		pushInfos, err := json.Marshal(c.PushInfos)
		pullInfos, err := json.Marshal(c.PullInfos)
		param := make(map[string]interface{})
		param["push_infos"] = string(pushInfos)
		param["pull_infos"] = string(pullInfos)
		err = tx.Model(&live_schedule.TrafficOff{}).Where("id=?", c.Id).Updates(&param).Error
		if err != nil {
			tx.Rollback()
			e.Log.Errorf("update TrafficOff error:%s \r\n", err)
			return err
		}

		cortexTaskService := CortexTaskService{}
		cortexTaskService.Log = e.Log
		cortexTaskService.Orm = e.Orm
		for k, v := range cortexTasks {
			v := v
			taskIds = append(taskIds, v.OutNodeId)
			key := strconv.Itoa(v.CortexOutputId) + models.Delimiter + strconv.Itoa(v.MediaProcessId) + models.Delimiter + strconv.Itoa(v.TrafficOffId)
			pushInfo, ok1 := pushInfoMap[key]
			//如果存在，并且没有转推任务直接保存
			if ok1 && len(pushInfo.PushUrl) > 0 && len(v.TaskId) <= 0 {
				*(&cortexTasks[k].PushUrl) = pushInfo.PushUrl
			}

			//如果存在，并且有转推任务 需要先清除任务
			if ok1 && len(pushInfo.PushUrl) > 0 && len(v.TaskId) > 0 && v.PushUrl != pushInfo.PushUrl {
				err = streamHubService.StopStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.TaskId})
				if err != nil {
					tx.Rollback()
					e.Log.Errorf("StopStreamHub error:%s \r\n", err)
					return err
				}
				err = streamHubService.DeleteStreamHub(&StreamHubDto.StreamHubReq{TenantId: v.TenantId, Id: v.TaskId})
				if err != nil {
					tx.Rollback()
					e.Log.Errorf("DeleteStreamHub error:%s \r\n", err)
					return err
				}
				scheduleIds = append(scheduleIds, v.TaskId)
				*(&cortexTasks[k].TaskId) = ""
				*(&cortexTasks[k].PushUrl) = pushInfo.PushUrl
			}

			pullInfo, ok2 := pullInfoMap[key]
			if ok2 && len(pullInfo.PullUrl) > 0 {
				arr := strings.Split(pullInfo.PullUrl, "/")
				if len(arr) > 1 {
					index := strings.Index(arr[1], ".")
					sub := arr[1][0:index]
					//如果编辑自定义参数且任务是启用状态需要调用编辑自定义参数接口 && channelBasic.ChannelState != 2
					if len(v.TaskId) > 0 {
						err = streamHubService.EditRefinePlayUrl(&StreamHubDto.EditRefinePlayUrlReq{TenantId: c.TenantId, Id: v.OutNodeId, RedefinePlayUrlEnable: "1", RedefinePlayUrlParams: StreamHubDto.RedefinePlayUrlParams{Path: arr[0], Sub: sub}})
						if err != nil {
							tx.Rollback()
							e.Log.Errorf("EditRefinePlayUrl error:%s \r\n", err)
							return err
						}
					}
				}
				*(&cortexTasks[k].CustomOutUrl) = pullInfo.PullUrl
			}
		}
		if len(channelTasks) > 0 {
			err = tx.Save(&channelTasks).Error
			if err != nil {
				tx.Rollback()
				e.Log.Errorf("save ChannelTask error:%s \r\n", err)
				return err
			}
		}
		if len(cortexTasks) > 0 {
			err = tx.Save(&cortexTasks).Error
			if err != nil {
				tx.Rollback()
				e.Log.Errorf("save cortexTasks error:%s \r\n", err)
				return err
			}
		}

	}

	// 自定义转成非自定义
	if originIsCustom == 1 && c.IsCustom == 2 {
		for _, v := range channelTasks {
			err = streamHubService.EditRefinePlayUrl(&StreamHubDto.EditRefinePlayUrlReq{TenantId: c.TenantId, Id: v.OutNodeId, RedefinePlayUrlEnable: "2", RedefinePlayUrlParams: StreamHubDto.RedefinePlayUrlParams{Path: "", Sub: ""}})
			if err != nil {
				tx.Rollback()
				e.Log.Errorf("EditRefinePlayUrl error:%s \r\n", err)
				return err
			}
		}
	}
	if len(scheduleIds) > 0 {
		//更新任务记录结束时间
		param := make(map[string]interface{})
		param["end_time"] = now
		err = tx.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id in ? and end_time = 0", scheduleIds).Updates(&param).Error
		if err != nil {
			e.Log.Errorf("Updates ScheduleRecord error:%s \r\n", err)
			tx.Rollback()
			return err
		}
	}

	//处理黑白名单
	if isDiff {
		cortexTasks := make([]live_schedule.CortexTask, 0, 5)
		err = tx.Model(live_schedule.CortexTask{}).Where("channel_id=? and traffic_off_id=?", c.ChannelId, c.Id).Find(&cortexTasks).Error
		if err != nil {
			tx.Rollback()
			e.Log.Errorf("get CortexTask error:%s \r\n", err)
			return err
		}
		for _, v := range cortexTasks {
			v := v
			taskIds = append(taskIds, v.OutNodeId)
		}
		channelTransCodeTasks := make([]live_schedule.ChannelTransCodeTask, 0, 5)
		err = tx.Model(live_schedule.ChannelTransCodeTask{}).Where("channel_id=? and traffic_off_id=?", c.ChannelId, c.Id).Find(&channelTransCodeTasks).Error
		if err != nil {
			tx.Rollback()
			e.Log.Errorf("get ChannelTransCodeTask error:%s \r\n", err)
			return err
		}
		for _, v := range channelTransCodeTasks {
			v := v
			taskIds = append(taskIds, v.OutNodeId)
		}
		e.ProcessBlackWhite(taskIds, data)
	}

	c.Id = data.Id
	c.NodeId = data.NodeId
	tx.Commit()
	return nil
}

func (e *ChannelTaskService) RemoveTrafficOff(d *dto.TrafficOffDeleteReq) error {
	var data live_schedule.TrafficOff

	db := e.Orm.Model(&data).Delete(&data, d.Id)
	if err := db.Error; err != nil {
		e.Log.Errorf("Service RemoveTrafficOff error:%s \r\n", err)
		return err
	}
	return nil
}

func (e *ChannelTaskService) GetTrafficOff(d *dto.TrafficOffGetReq) (*live_schedule.TrafficOffModel, error) {
	var model live_schedule.TrafficOffModel
	var data live_schedule.TrafficOff

	sourceBasicService := SourceBasicService{}
	sourceBasicService.Orm = e.Orm
	sourceBasicService.Log = e.Log
	//获取下车点基本信息
	err := e.Orm.Model(&data).
		First(&data, d.Id).Error
	if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
		e.Log.Errorf("Service GetTrafficOff error:%s \r\n", err)
		return nil, err
	}
	if data.Id <= 0 {
		return nil, nil
	}
	model.TrafficOff = data

	//获取媒体处理信息
	mediaProcess := make([]live_schedule.MediaProcess, 0, 5)
	err = e.Orm.Model(live_schedule.MediaProcess{}).Where("tenant_id =? and channel_id=?", data.TenantId, data.ChannelId).Find(&mediaProcess).Error
	if err != nil {
		e.Log.Errorf("Service GetMediaProcess error:%s \r\n", err)
		return nil, err
	}

	//获取所有上车点
	upIds := make([]int, 0, 5)
	err = e.Orm.Model(&live_schedule.UpMedia{}).Select("up_id").Find(&upIds).Error
	if err != nil {
		e.Log.Errorf("UpMedia error:%s \r\n", err)
		return nil, err
	}
	sourceIds := make([]int, 0, 5)
	ups := make([]live_schedule.TrafficOn, 0, 5)
	upMap := make(map[int]live_schedule.TrafficOn)
	err = e.Orm.Model(&live_schedule.TrafficOn{}).Where("tenant_id =? and channel_id=?", data.TenantId, data.ChannelId).Find(&ups).Error
	if err != nil {
		e.Log.Errorf("get TrafficOn error:%s \r\n", err)
		return nil, err
	}
	for _, v := range ups {
		v := v
		upMap[v.Id] = v
		sourceIds = append(sourceIds, v.SourceId)
	}

	//获取信源对应的转码任务
	transCodeTask := make([]live_schedule.SourceTemplate, 0, 5)
	err = e.Orm.Model(&live_schedule.SourceTemplate{}).Where("source_id in ?", sourceIds).Find(&transCodeTask).Error
	if err != nil {
		e.Log.Errorf("get TransCodeTask error:%s \r\n", err)
		return nil, err
	}

	transMap := make(map[string]live_schedule.SourceTemplate)
	for _, v := range transCodeTask {
		v := v
		transMap[v.TransCodeOutputAddress] = v
	}
	//获取信源
	sourceMap := make(map[int]live_schedule.SourceBasic)
	var count int64
	sources, err := sourceBasicService.GetPage(&dto.SourceBasicGetPageReq{SourceIds: sourceIds, Pagination: commonDto.Pagination{PageIndex: 1, PageSize: 10000}, Kind: 3}, &count)
	if err != nil {
		e.Log.Errorf("get sourceBasic error:%s \r\n", err)
		return nil, err
	}
	for _, v := range *sources {
		v := v
		sourceMap[v.Basic.Id] = *v.Basic
	}
	infos := make([]live_schedule.NodeInfo, 0, 5)
	now := time.Now().Unix()
	for _, m := range mediaProcess {
		if m.MediaType == 1 {
			//获取下车点的通道信息
			var tasks []live_schedule.ChannelTask
			err = e.Orm.Model(live_schedule.ChannelTask{}).Where("tenant_id =? and channel_id=? and traffic_off_id=?", data.TenantId, data.ChannelId, data.Id).Find(&tasks).Error
			if err != nil {
				e.Log.Errorf("Service GetTrafficOff error:%s \r\n", err)
				return nil, err
			}
			for _, task := range tasks {
				sourceIds = append(sourceIds, task.SourceId)
			}

			for _, task := range tasks {

				//自动调度内网忽略
				if task.Area == 2 {
					continue
				}

				//转码任务被删除直接跳过
				up := upMap[task.TrafficOnId]
				if up.StreamType == 2 {
					_, ok := transMap[up.TransUdpUrl]
					if !ok {
						continue
					}
				}
				info := live_schedule.NodeInfo{}
				if sourceMap[task.SourceId].ProtocolType != 5 && task.TaskState != 3 {
					if sourceMap[task.SourceId].SourceState == 1 {
						task.TaskState = 1
						info.NodeState = 1
					} else {
						task.TaskState = 2
						info.NodeState = 2
					}
				} else {
					if task.TaskState == 1 && task.StateUpdateTime+10 < now {
						info.NodeState = 1
					} else {
						info.NodeState = 2
					}
				}
				if task.TaskState == 3 {
					info.NodeState = 3
				}
				e.Log.Trace("下车点任务状态：", task.TaskState, task.OutUrl)
				info.UpId = task.TrafficOnId
				info.SourceId = task.SourceId
				info.NodeUrl = task.OutUrl
				info.InNodeId = task.InNodeId
				info.OutNodeId = task.OutNodeId
				info.OutputStreamId = task.OutputStreamId
				info.CustomUrl = task.CustomOutUrl
				if up.StreamType == 2 {
					t := transMap[up.TransUdpUrl]
					info.TransCodeName = t.TaskName
				}
				info.UpId = up.Id
				infos = append(infos, info)
			}
			model.NodeInfos = infos
		} else if m.MediaType == 2 {
			//获取下车点的通道信息
			var tasks []live_schedule.CortexTask
			outIds := make([]int, 0, 5)
			err = e.Orm.Model(live_schedule.CortexTask{}).Where("tenant_id =? and channel_id=? and traffic_off_id=? and task_id !=''", data.TenantId, data.ChannelId, data.Id).Find(&tasks).Error
			if err != nil {
				e.Log.Errorf("Service GetCortexTask error:%s \r\n", err)
				return nil, err
			}
			for _, task := range tasks {
				sourceIds = append(sourceIds, task.SourceId)
				outIds = append(outIds, task.CortexOutputId)
			}

			//获取输出名
			outs := make([]live_schedule.CortexOutput, 0, 5)
			outMap := make(map[int]string)
			err = e.Orm.Model(live_schedule.CortexOutput{}).Where("id in ?", outIds).Find(&outs).Error
			if err != nil {
				e.Log.Errorf("get CortexOutput error:%s \r\n", err)
				return nil, err
			}
			for _, v := range outs {
				outMap[v.Id] = v.OutName
			}

			for _, task := range tasks {
				//转码任务被删除直接跳过
				up := upMap[task.TrafficOnId]
				if up.StreamType == 2 {
					_, ok := transMap[up.TransUdpUrl]
					if !ok {
						continue
					}
				}
				info := live_schedule.NodeInfo{}

				//自动调度内网忽略
				if task.Area == 2 {
					continue
				}
				//如果信源是udp 状态特殊处理
				if sourceMap[task.SourceId].ProtocolType != 5 {
					if sourceMap[task.SourceId].SourceState == 1 {
						//task.TaskState = 1
						info.NodeState = 1
					} else {
						//task.TaskState = 2
						info.NodeState = 2
					}
				} else {
					info.NodeState = task.TaskState
				}
				if task.TaskState == 3 {
					info.NodeState = 3
				}
				info.SourceId = task.SourceId
				info.NodeState = task.TaskState
				info.NodeUrl = task.OutUrl
				info.InNodeId = task.InNodeId
				info.OutNodeId = task.OutNodeId
				info.OutputStreamId = task.OutputStreamId
				outName := outMap[task.CortexOutputId]
				info.OutputName = outName
				info.CustomUrl = task.CustomOutUrl

				if up.StreamType == 2 {
					t := transMap[up.TransUdpUrl]
					info.TransCodeName = t.TaskName
				}
				info.UpId = up.Id
				infos = append(infos, info)
				outIds = append(outIds, task.CortexOutputId)
			}
			model.NodeInfos = infos
		} else if m.MediaType == 4 {
			//获取下车点的通道信息
			var tasks []live_schedule.ChannelTransCodeTask
			err = e.Orm.Model(live_schedule.ChannelTransCodeTask{}).Where("tenant_id =? and channel_id=? and traffic_off_id=?", data.TenantId, data.ChannelId, data.Id).Find(&tasks).Error
			if err != nil {
				e.Log.Errorf("Service GetTrafficOff error:%s \r\n", err)
				return nil, err
			}
			for _, task := range tasks {
				sourceIds = append(sourceIds, task.SourceId)
			}

			for _, task := range tasks {

				//自动调度内网忽略
				if task.Area == 2 {
					continue
				}

				//转码任务被删除直接跳过
				up := upMap[task.TrafficOnId]
				if up.StreamType == 2 {
					_, ok := transMap[up.TransUdpUrl]
					if !ok {
						continue
					}
				}
				info := live_schedule.NodeInfo{}
				if sourceMap[task.SourceId].ProtocolType != 5 {
					if sourceMap[task.SourceId].SourceState == 1 {
						task.TaskState = 1
						info.NodeState = 1
					} else {
						task.TaskState = 2
						info.NodeState = 2
					}
				} else {
					if task.TaskState == 1 && task.StateUpdateTime+10 < now {
						info.NodeState = 1
					} else {
						info.NodeState = 2
					}
				}
				e.Log.Trace("下车点任务状态：", task.TaskState, task.OutUrl)
				info.UpId = task.TrafficOnId
				info.SourceId = task.SourceId
				info.NodeUrl = task.OutUrl
				info.InNodeId = task.InNodeId
				info.OutNodeId = task.OutNodeId
				info.OutputStreamId = task.OutputStreamId
				info.CustomUrl = task.CustomOutUrl
				if up.StreamType == 2 {
					t := transMap[up.TransUdpUrl]
					info.TransCodeName = t.TaskName
				}
				info.UpId = up.Id
				infos = append(infos, info)
			}
			model.NodeInfos = infos
		} else if m.MediaType == 3 {
			//获取下车点的通道信息
			var tasks []live_schedule.AuditTask
			err = e.Orm.Model(live_schedule.AuditTask{}).Where("tenant_id =? and channel_id=? and traffic_off_id=?", data.TenantId, data.ChannelId, data.Id).Find(&tasks).Error
			if err != nil {
				e.Log.Errorf("Service GetTrafficOff error:%s \r\n", err)
				return nil, err
			}
			for _, task := range tasks {
				sourceIds = append(sourceIds, task.SourceId)
			}

			for _, task := range tasks {
				//自动调度内网忽略
				if task.Area == 2 {
					continue
				}

				info := live_schedule.NodeInfo{}
				if sourceMap[task.SourceId].ProtocolType != 5 {
					if sourceMap[task.SourceId].SourceState == 1 {
						task.TaskState = 1
						info.NodeState = 1
					} else {
						task.TaskState = 2
						info.NodeState = 2
					}
				} else {
					info.NodeState = task.TaskState
				}

				e.Log.Trace("下车点任务状态：", task.TaskState)
				info.UpId = task.TrafficOnId
				info.NodeState = task.TaskState
				info.SourceId = task.SourceId
				info.NodeUrl = task.OutUrl
				info.InNodeId = task.InNodeId
				info.OutNodeId = task.OutNodeId
				info.OutputStreamId = task.OutputStreamId
				info.CustomUrl = task.CustomOutUrl
				infos = append(infos, info)
			}
			model.NodeInfos = infos
		} else if m.MediaType == 6 {
			//获取下车点的通道信息
			var tasks []live_schedule.WaterMarkTransTask
			err = e.Orm.Model(live_schedule.WaterMarkTransTask{}).Where("tenant_id =? and channel_id=? and traffic_off_id=?", data.TenantId, data.ChannelId, data.Id).Find(&tasks).Error
			if err != nil {
				e.Log.Errorf("Service GetTrafficOff error:%s \r\n", err)
				return nil, err
			}
			for _, task := range tasks {
				sourceIds = append(sourceIds, task.SourceId)
			}

			for _, task := range tasks {

				//自动调度内网忽略
				if task.Area == 2 {
					continue
				}

				//转码任务被删除直接跳过
				up := upMap[task.TrafficOnId]
				if up.StreamType == 2 {
					_, ok := transMap[up.TransUdpUrl]
					if !ok {
						continue
					}
				}
				info := live_schedule.NodeInfo{}
				if sourceMap[task.SourceId].ProtocolType != 5 {
					if sourceMap[task.SourceId].SourceState == 1 {
						task.TaskState = 1
						info.NodeState = 1
					} else {
						task.TaskState = 2
						info.NodeState = 2
					}
				} else {
					if task.TaskState == 1 && task.StateUpdateTime+10 < now {
						info.NodeState = 1
					} else {
						info.NodeState = 2
					}
				}
				e.Log.Trace("下车点任务状态：", task.TaskState, task.OutUrl)
				info.UpId = task.TrafficOnId
				info.SourceId = task.SourceId
				info.NodeUrl = task.OutUrl
				info.InNodeId = task.InNodeId
				info.OutNodeId = task.OutNodeId
				info.OutputStreamId = task.OutputStreamId
				info.CustomUrl = task.CustomOutUrl
				if up.StreamType == 2 {
					t := transMap[up.TransUdpUrl]
					info.TransCodeName = t.TaskName
				}
				info.UpId = up.Id
				infos = append(infos, info)
			}
			model.NodeInfos = infos
		}
	}

	//获取黑白名单
	hlsBlackWhites := make([]live_schedule.HlsBlackWhite, 0, 5)
	err = e.Orm.Model(&live_schedule.HlsBlackWhite{}).Where("off_id = ?", d.Id).Find(&hlsBlackWhites).Error
	if err != nil {
		e.Log.Errorf("get HlsBlackWhite error:%s \r\n", err)
		return nil, err
	}
	model.TrafficOff.IPAddresses = hlsBlackWhites
	return &model, nil
}

func (e *ChannelTaskService) SaveWorkflow(c *dto.SaveWorkflowReq, basic live_schedule.ChannelBasic) error {

	defer func() {
		if r := recover(); r != nil {
			e.Log.Errorf("SaveWorkflow error:%s \r\n", r)
			err := e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", c.ChannelId).Update("channel_state", basic.ChannelState).Error
			if err != nil {
				e.Log.Errorf("update ChannelBasic error:%s \r\n", err)
			}
		}
	}()

	var wg sync.WaitGroup
	liveService := &liveService.LiveBasic{}
	streamHubService := &streamHubService.StreamHubService{}
	baiduService := &baiduService.BaiduService{}
	streamHubService.Log = e.Log
	liveService.Log = e.Log
	baiduService.Log = e.Log
	cortexTaskService := &CortexTaskService{}
	cortexTaskService.Log = e.Log
	cortexTaskService.Orm = e.Orm

	liveBasicService := SourceBasicService{}
	liveBasicService.Log = e.Log
	liveBasicService.Orm = e.Orm

	auditTaskService := AuditTaskService{}
	auditTaskService.Log = e.Log
	auditTaskService.Orm = e.Orm

	recordTaskService := RecordTaskService{}
	recordTaskService.Log = e.Log
	recordTaskService.Orm = e.Orm

	convertTaskService := ConvertTaskService{}
	convertTaskService.Log = e.Log
	convertTaskService.Orm = e.Orm

	transCodeTaskService := TransCodeTaskService{}
	transCodeTaskService.Log = e.Log
	transCodeTaskService.Orm = e.Orm

	waterMarkTaskService := WaterMarkTaskService{}
	waterMarkTaskService.Log = e.Log
	waterMarkTaskService.Orm = e.Orm

	//原始协转任务配置关系key
	originKeys := make(map[string]live_schedule.ChannelTask)
	//原始协转任务配置包括自动调
	originAllKeys := make(map[string]live_schedule.ChannelTask)
	//原始矩阵任务配置包括自动调
	originCortexAllKeys := make(map[string]live_schedule.CortexTask)
	//原始软矩阵任务配置关系key
	originCortexKeys := make(map[string]live_schedule.CortexTask)
	//原始审核任务配置关系key
	originAuditKeys := make(map[string]live_schedule.AuditTask)
	//原始审核任务配置包括自动调
	originAuditKeysAllKeys := make(map[string]live_schedule.AuditTask)
	//原始转码任务配置关系key
	originTransCodeKeys := make(map[string]live_schedule.ChannelTransCodeTask)
	//原始转码任务配置包括自动调
	originTransCodeAllKeys := make(map[string]live_schedule.ChannelTransCodeTask)
	//原始录制任务配置关系key
	originRecordKeys := make(map[string]live_schedule.SourceRecordTask)
	//原始水印任务配置关系key
	originWatermarkKeys := make(map[string]live_schedule.WaterMarkTransTask)
	//原始水印任务配置包括自动调
	originWatermarkAllKeys := make(map[string]live_schedule.WaterMarkTransTask)
	//需要添加的协转任务配置
	addConvertItems := make([]live_schedule.ChannelTask, 0, 5)
	//需要添加的软矩阵任务配置
	addCortexItems := make([]live_schedule.CortexTask, 0, 5)
	//需要删除汇聚协转的配置
	delConvertItems := make([]live_schedule.ChannelTask, 0, 5)
	//需要删除汇聚协转的配置
	delCortexItems := make([]live_schedule.CortexTask, 0, 5)
	//需要删除录制任务的配置
	delRecordItems := make([]live_schedule.SourceRecordTask, 0, 5)
	//山车点id
	upIds := make([]int, 0, 5)
	//媒体处理id
	mediaIds := make([]int, 0, 5)
	//下车点id
	downIds := make([]int, 0, 5)
	//输出名
	cortexOutputs := make([]live_schedule.CortexOutput, 0, 5)

	//根据通道id获取所有任务
	originTasks := make([]live_schedule.ChannelTask, 0, 5)
	//软矩阵任务
	cortexTasks := make([]live_schedule.CortexTask, 0, 5)
	//正在使用的软矩阵任务
	useCortexTaskMap := make(map[string][]live_schedule.CortexTask)

	//审核原始任务
	originAuditTask := make([]live_schedule.AuditTask, 0, 5)
	//需要添加的审核任务配置
	addAuditItems := make([]live_schedule.AuditTask, 0, 5)
	//需要删除审核协转任务
	delAuditItems := make([]live_schedule.AuditTask, 0, 5)

	//转码原始任务
	originTransCodeTask := make([]live_schedule.ChannelTransCodeTask, 0, 5)
	//需要添加的转码任务配置
	addTransCodeItems := make([]live_schedule.ChannelTransCodeTask, 0, 5)
	//需要删除的转码任务
	delTransCodeItems := make([]live_schedule.ChannelTransCodeTask, 0, 5)
	//原始录制任务
	originRecordTask := make([]live_schedule.SourceRecordTask, 0, 5)
	//需要添加的录制任务配置
	addRecordItems := make([]live_schedule.SourceRecordTask, 0, 5)
	//原始水印任务
	originWatermarkTask := make([]live_schedule.WaterMarkTransTask, 0, 5)
	//需要添加的水印任务配置
	addWatermarkItems := make([]live_schedule.WaterMarkTransTask, 0, 5)
	//需要删除水印任务
	delWatermarkItems := make([]live_schedule.WaterMarkTransTask, 0, 5)
	sourcesMap := make(map[int]live_schedule.SourceBasic)

	workTotalNum := 0
	var partNum int
	now := time.Now().Unix()

	////获取通道基本信息
	//var basic live_schedule.ChannelBasic
	//e.Orm.Model(live_schedule.ChannelBasic{}).Where("tenant_id =? and id=?", c.TenantId, c.ChannelId).Find(&basic)

	count := e.CheckScheduleCount(c.TenantId)
	e.Log.Trace("当前调度任务数:", count)
	recordCount := 0
	for _, v := range c.WorkflowMedia {
		if v.MediaProcess.MediaType == 1 {
			err := e.Orm.Model(live_schedule.ChannelTask{}).Where("tenant_id =? and channel_id=?", c.TenantId, c.ChannelId).Find(&originTasks).Error
			if err != nil {
				e.Log.Errorf("get ChannelTask error:%s \r\n", err)
				return err
			}
		} else if v.MediaProcess.MediaType == 2 {
			err := e.Orm.Model(live_schedule.CortexTask{}).Where("tenant_id =? and channel_id=?", c.TenantId, c.ChannelId).Find(&cortexTasks).Error
			if err != nil {
				e.Log.Errorf("get cortexTasks error:%s \r\n", err)
				return err
			}

			err = e.Orm.Model(live_schedule.CortexOutput{}).Where("tenant_id =? and channel_id=? and media_process_id=?", c.TenantId, c.ChannelId, c.WorkflowMedia[0].MediaProcess.Id).Find(&cortexOutputs).Error
			if err != nil {
				e.Log.Errorf("get CortexOutput error:%s \r\n", err)
				return err
			}

		} else if v.MediaProcess.MediaType == 3 {
			err := e.Orm.Model(live_schedule.AuditTask{}).Where("tenant_id =? and channel_id=?", c.TenantId, c.ChannelId).Find(&originAuditTask).Error
			if err != nil {
				e.Log.Errorf("get AuditTask error:%s \r\n", err)
				return err
			}
		} else if v.MediaProcess.MediaType == 4 {
			err := e.Orm.Model(live_schedule.ChannelTransCodeTask{}).Where("tenant_id =? and channel_id=?", c.TenantId, c.ChannelId).Find(&originTransCodeTask).Error
			if err != nil {
				e.Log.Errorf("get ChannelTransCodeTask error:%s \r\n", err)
				return err
			}
		} else if v.MediaProcess.MediaType == 5 {
			recordCount++
			if recordCount > 1 {
				return errors.New("不能重复创建录制节点")
			}
			err := e.Orm.Model(live_schedule.SourceRecordTask{}).Where("tenant_id =? and channel_id=?", c.TenantId, c.ChannelId).Find(&originRecordTask).Error
			if err != nil {
				e.Log.Errorf("get SourceRecordTask error:%s \r\n", err)
				return err
			}
		} else if v.MediaProcess.MediaType == 6 {
			err := e.Orm.Model(live_schedule.WaterMarkTransTask{}).Where("tenant_id =? and channel_id=?", c.TenantId, c.ChannelId).Find(&originWatermarkTask).Error
			if err != nil {
				e.Log.Errorf("get WaterMarkTransTask error:%s \r\n", err)
				return err
			}
		}
	}

	//获取正在输出的矩阵任务
	usingCortexMap := make(map[int]int)
	//标记输出点是否已经配置
	enableCortexMap := make(map[string]int)
	for _, v := range cortexTasks {
		v := v
		key := strconv.Itoa(v.TrafficOnId) + models.Delimiter + strconv.Itoa(v.SourceId) + models.Delimiter + strconv.Itoa(v.CortexOutputId) + models.Delimiter + strconv.Itoa(v.TrafficOffId)
		if len(v.TaskId) > 0 {
			tasks, ok := useCortexTaskMap[key]
			if !ok {
				tasks = make([]live_schedule.CortexTask, 0, 5)
			}
			tasks = append(tasks, v)
			useCortexTaskMap[key] = tasks
			usingCortexMap[v.CortexOutputId] = v.SourceId
		}
		enableCortexMap[key] = v.IsEnable
	}
	e.Log.Trace("存储配置的输出点：", enableCortexMap)

	//获取现有山车点下车点媒体处理的配置id
	for _, v := range c.WorkflowMedia {
		upIds = append(upIds, v.TransferUpPointId...)
		downIds = append(downIds, v.TransferDownPointId...)
		mediaIds = append(mediaIds, v.TransferMediaPointId)
	}

	trafficOns := make([]live_schedule.TrafficOn, 0, 5)
	err := e.Orm.Model(live_schedule.TrafficOn{}).Where("id in ?", upIds).Find(&trafficOns).Error
	if err != nil {
		e.Log.Errorf("get TrafficOn error:%s \r\n", err)
		return err
	}

	//存储山车点的信源id 是组信源还是单信源
	upMap := make(map[int]live_schedule.TrafficOn)
	//存储上车点组信源的id
	isGroupsIds := make([]int, 0, 5)
	for _, v := range trafficOns {
		v := v
		upMap[v.Id] = v
		if v.IsGroup == 1 {
			isGroupsIds = append(isGroupsIds, v.SourceId)
		}
	}
	e.Log.Trace("上车点存储的信源", upMap)

	cortexOutputMap := make(map[int]live_schedule.CortexOutput)
	for _, v := range cortexOutputs {
		v := v
		cortexOutputMap[v.Id] = v
	}

	//获取软矩阵的输出名
	outputs := make([]live_schedule.CortexOutput, 0, 5)
	err = e.Orm.Model(live_schedule.CortexOutput{}).Where("media_process_id in ?", mediaIds).Find(&outputs).Error
	if err != nil {
		e.Log.Errorf("get CortexOutput error:%s \r\n", err)
		return err
	}
	outputMap := make(map[int]live_schedule.CortexOutput)
	for _, v := range outputs {
		v := v
		outputMap[v.Id] = v
	}

	//如果是组信源id需要获取组下面的所有单信源
	relations := make([]live_schedule.SourceGroupRelation, 0, 5)
	err = e.Orm.Model(live_schedule.SourceGroupRelation{}).Where("group_id in ?", isGroupsIds).Find(&relations).Error
	if err != nil {
		e.Log.Errorf("ChannelTaskService SourceGroupRelation error:%s \r\n", err)
		return err
	}
	e.Log.Trace("组信源id需要获取组下面的所有单信源", relations)
	//return err
	//存储上车点和组信源下面的信源的映射关系
	relationMap := make(map[int][]int)
	for _, v := range trafficOns {
		if v.IsGroup == 1 {
			for _, r := range relations {
				relation, ok := relationMap[v.Id]
				if !ok {
					relation = make([]int, 0, 5)
				}
				relation = append(relation, r.SourceId)
				relationMap[v.Id] = relation
			}
		}
	}
	e.Log.Trace("组信源列表：", relationMap)

	//获取黑白名单
	hlsBlackWhites := make([]live_schedule.HlsBlackWhite, 0, 5)
	e.Orm.Model(&live_schedule.HlsBlackWhite{}).Where("off_id in ?", downIds).Find(&hlsBlackWhites)
	hlsBlackWhiteMap := make(map[int][]live_schedule.HlsBlackWhite)
	for _, v := range hlsBlackWhites {
		v := v
		list, ok := hlsBlackWhiteMap[v.OffId]
		if !ok {
			list = make([]live_schedule.HlsBlackWhite, 0, 5)
		}
		list = append(list, v)
		hlsBlackWhiteMap[v.OffId] = list
	}

	//获取所有下车点信息
	trafficOffs := make([]live_schedule.TrafficOff, 0, 5)
	e.Orm.Model(live_schedule.TrafficOff{}).Where("id in ?", downIds).Find(&trafficOffs)
	offMap := make(map[int]live_schedule.TrafficOff)
	for _, v := range trafficOffs {
		v := v
		//设置黑白名单
		list := hlsBlackWhiteMap[v.Id]
		v.IPAddresses = list
		offMap[v.Id] = v
	}

	//获取转推和拉流地址
	pushMap := make(map[string]dto.PushInfo)
	pullMap := make(map[string]dto.PullInfo)
	pushCortexMap := make(map[string]dto.PushInfo)
	pullCortexMap := make(map[string]dto.PullInfo)

	if len(c.PushInfos) > 0 {
		for _, v := range c.PushInfos {
			v := v
			key := strconv.Itoa(v.SourceId) + models.Delimiter + strconv.Itoa(v.UpId) + models.Delimiter + strconv.Itoa(v.MediaId) + models.Delimiter + strconv.Itoa(v.OffId)
			pushMap[key] = v
			key1 := strconv.Itoa(v.CortexOutId) + models.Delimiter + strconv.Itoa(v.MediaId) + models.Delimiter + strconv.Itoa(v.OffId)
			pushCortexMap[key1] = v
		}
	}
	if len(c.PullInfos) > 0 {
		for _, v := range c.PullInfos {
			v := v
			key := strconv.Itoa(v.SourceId) + models.Delimiter + strconv.Itoa(v.UpId) + models.Delimiter + strconv.Itoa(v.MediaId) + models.Delimiter + strconv.Itoa(v.OffId)
			pullMap[key] = v
			key1 := strconv.Itoa(v.CortexOutId) + models.Delimiter + strconv.Itoa(v.MediaId) + models.Delimiter + strconv.Itoa(v.OffId)
			pullCortexMap[key1] = v
		}
	}

	//汇聚协转和软矩阵
	if len(c.WorkflowMedia) > 0 {
		e.Log.Trace("开始处理调度任务")
		for _, v := range c.WorkflowMedia {
			//组装原始通道配置关系key 和存储特殊处理逻辑
			if v.MediaProcess.MediaType == 1 {
				for _, v := range originTasks {
					v := v
					key := strconv.Itoa(v.SourceId) + models.Delimiter + strconv.Itoa(v.TrafficOnId) + models.Delimiter + strconv.Itoa(v.MediaId) + models.Delimiter + strconv.Itoa(v.TrafficOffId)
					originKeys[key] = v
					key1 := key + models.Delimiter + strconv.Itoa(v.Area)
					originAllKeys[key1] = v
				}
			} else if v.MediaProcess.MediaType == 2 {
				for _, v := range cortexTasks {
					v := v
					key := strconv.Itoa(v.SourceId) + models.Delimiter + strconv.Itoa(v.TrafficOnId) + models.Delimiter + strconv.Itoa(v.MediaProcessId) + models.Delimiter + strconv.Itoa(v.TrafficOffId) + models.Delimiter + strconv.Itoa(v.CortexOutputId)
					originCortexKeys[key] = v
					key1 := key + models.Delimiter + strconv.Itoa(v.Area)
					originCortexAllKeys[key1] = v
				}
			} else if v.MediaProcess.MediaType == 3 {
				//获取审核任务
				for _, v := range originAuditTask {
					v := v
					key := strconv.Itoa(v.SourceId) + models.Delimiter + strconv.Itoa(v.TrafficOnId) + models.Delimiter + strconv.Itoa(v.MediaId) + models.Delimiter + strconv.Itoa(v.TrafficOffId)
					originAuditKeys[key] = v
					key1 := key + models.Delimiter + strconv.Itoa(v.Area)
					originAuditKeysAllKeys[key1] = v
				}
			} else if v.MediaProcess.MediaType == 4 {
				//获取转码任务
				for _, v := range originTransCodeTask {
					v := v
					key := strconv.Itoa(v.SourceId) + models.Delimiter + strconv.Itoa(v.TrafficOnId) + models.Delimiter + strconv.Itoa(v.MediaId) + models.Delimiter + strconv.Itoa(v.TrafficOffId)
					originTransCodeKeys[key] = v
					key1 := key + models.Delimiter + strconv.Itoa(v.Area)
					originTransCodeAllKeys[key1] = v
				}
			} else if v.MediaProcess.MediaType == 5 {
				for _, v := range originRecordTask {
					v := v
					key := strconv.Itoa(v.SourceId) + models.Delimiter + strconv.Itoa(v.TrafficOnId) + models.Delimiter + strconv.Itoa(v.MediaId)
					originRecordKeys[key] = v
				}
			} else if v.MediaProcess.MediaType == 6 {
				for _, v := range originWatermarkTask {
					v := v
					key := strconv.Itoa(v.SourceId) + models.Delimiter + strconv.Itoa(v.TrafficOnId) + models.Delimiter + strconv.Itoa(v.MediaId) + models.Delimiter + strconv.Itoa(v.TrafficOffId)
					originWatermarkKeys[key] = v
					key1 := key + models.Delimiter + strconv.Itoa(v.Area)
					originWatermarkAllKeys[key1] = v
				}
			}
		}

		//组装汇聚协转入参关系参数
		inputTaskKeys := make(map[string]struct{})
		//组装软矩阵关系参数
		inputCortexTaskKeys := make(map[string]struct{})
		//组装审核关系参数
		inputAuditTaskKeys := make(map[string]struct{})
		//组装录制关系参数
		inputRecordTaskKeys := make(map[string]struct{})
		//组装转码关系参数
		inputTransCodeTaskKeys := make(map[string]struct{})
		//组装水印关系参数
		inputWatermarkTaskKeys := make(map[string]struct{})

		//auditId := 0
		sourcesIds := make([]int, 0, 5)
		for _, v := range c.WorkflowMedia {
			v := v
			for _, up := range v.TransferUpPointId {
				//遍历上车点
				tempSourcesIds := make([]int, 0, 5)
				//如果是信源组取出所有信源
				if upMap[up].IsGroup == 1 {
					tempSourcesIds = relationMap[up]
				} else {
					//单信源直接添加
					tempSourcesIds = append(tempSourcesIds, upMap[up].SourceId)
				}
				sourcesIds = append(sourcesIds, upMap[up].SourceId)
				e.Log.Trace("需要添加任务的信源id：", tempSourcesIds)

				//遍历信源
				for _, id := range tempSourcesIds {
					if v.MediaProcess.MediaType == 5 {
						key := strconv.Itoa(id) + models.Delimiter + strconv.Itoa(up) + models.Delimiter + strconv.Itoa(v.TransferMediaPointId)
						inputRecordTaskKeys[key] = struct{}{}
						//如果不存在就加入添加列表
						originTask, ok := originRecordKeys[key]
						if ok && len(originTask.RecordTaskId) > 0 {
							continue
						}
						task := live_schedule.SourceRecordTask{
							TenantId:    c.TenantId,
							ChannelId:   c.ChannelId,
							SourceId:    id,
							TrafficOnId: up,
							MediaId:     v.TransferMediaPointId,
							PushUrl:     originTask.PushUrl,
						}
						task.Id = originTask.Id
						task.CreateTime = now
						task.UpdateTime = now
						workTotalNum++
						addRecordItems = append(addRecordItems, task)
						continue
					}

					//遍历下车点
					for _, down := range v.TransferDownPointId {
						key := strconv.Itoa(id) + models.Delimiter + strconv.Itoa(up) + models.Delimiter + strconv.Itoa(v.TransferMediaPointId) + models.Delimiter + strconv.Itoa(down)
						off := offMap[down]
						pushInfo, pushOk := pushMap[key]
						pullInfo, pullOk := pullMap[key]
						if v.MediaProcess.MediaType == 1 {
							if off.Area == 3 && ((off.StreamType == 2 && off.TrafficOffType == 1) || off.TrafficOffType == 2 || off.TrafficOffType == 6) {
								key1 := key + models.Delimiter + "1"
								inputTaskKeys[key1] = struct{}{}
								key2 := key + models.Delimiter + "2"
								inputTaskKeys[key2] = struct{}{}
							} else {
								key3 := key + models.Delimiter + "0"
								inputTaskKeys[key3] = struct{}{}
							}

							originTask, ok := originKeys[key]
							if ok && len(originTask.TaskId) > 0 {
								continue
							}
							task := live_schedule.ChannelTask{
								TenantId:     c.TenantId,
								ChannelId:    c.ChannelId,
								SourceId:     id,
								TrafficOnId:  up,
								MediaId:      v.TransferMediaPointId,
								TrafficOffId: down,
								PushUrl:      originTask.PushUrl,
								CustomOutUrl: originTask.CustomOutUrl,
							}
							if pushOk {
								task.PushUrl = pushInfo.PushUrl
							}
							if pullOk {
								task.CustomOutUrl = pullInfo.PullUrl
							}
							task.Id = originTask.Id
							task.CreateTime = now
							task.UpdateTime = now
							workTotalNum++
							count++
							addConvertItems = append(addConvertItems, task)
						} else if v.MediaProcess.MediaType == 2 {
							for _, o := range cortexOutputs {
								key := strconv.Itoa(id) + models.Delimiter + strconv.Itoa(up) + models.Delimiter + strconv.Itoa(v.TransferMediaPointId) + models.Delimiter + strconv.Itoa(down) + models.Delimiter + strconv.Itoa(o.Id)
								cortexKey := strconv.Itoa(o.Id) + models.Delimiter + strconv.Itoa(v.TransferMediaPointId) + models.Delimiter + strconv.Itoa(down)

								if off.StreamType == 1 {
									err = json.Unmarshal([]byte(off.PushInfos), &c.PushInfos)
									for _, info := range c.PushInfos {
										key := strconv.Itoa(info.CortexOutId) + models.Delimiter + strconv.Itoa(info.MediaId) + models.Delimiter + strconv.Itoa(info.OffId)
										pushCortexMap[key] = info
									}
								}

								if off.StreamType == 2 && off.IsCustom == 1 {
									err = json.Unmarshal([]byte(off.PullInfos), &c.PullInfos)
									for _, info := range c.PullInfos {
										key := strconv.Itoa(info.CortexOutId) + models.Delimiter + strconv.Itoa(info.MediaId) + models.Delimiter + strconv.Itoa(info.OffId)
										pullCortexMap[key] = info
									}
								}
								//err = json.Unmarshal([]byte(off.PullInfos), &c.PullInfos)
								pushInfo, pushOk := pushCortexMap[cortexKey]
								pullInfo, pullOk := pullCortexMap[cortexKey]
								if off.Area == 3 && ((off.StreamType == 2 && off.TrafficOffType == 1) || off.TrafficOffType == 2 || off.TrafficOffType == 6) {
									key1 := key + models.Delimiter + "1"
									inputCortexTaskKeys[key1] = struct{}{}
									key2 := key + models.Delimiter + "2"
									inputCortexTaskKeys[key2] = struct{}{}
								} else {
									key3 := key + models.Delimiter + "0"
									inputCortexTaskKeys[key3] = struct{}{}
								}
								//inputCortexTaskKeys[key] = struct{}{}
								//如果不存任务在就加入添加列表
								//for s, _ := range originCortexKeys {
								//	e.Log.Trace("YYYYYYYYYYYYYYYYYYYYY：", s, key)
								//}
								originCortexTask, ok := originCortexKeys[key]
								task := live_schedule.CortexTask{
									TenantId:       c.TenantId,
									ChannelId:      c.ChannelId,
									SourceId:       id,
									TrafficOnId:    up,
									MediaProcessId: v.TransferMediaPointId,
									TrafficOffId:   down,
									CortexOutputId: o.Id,
									PushUrl:        originCortexTask.PushUrl,
									CustomOutUrl:   originCortexTask.CustomOutUrl,
								}
								enableKey := strconv.Itoa(up) + models.Delimiter + strconv.Itoa(id) + models.Delimiter + strconv.Itoa(o.Id) + models.Delimiter + strconv.Itoa(off.Id)

								if ok {
									//sourceId, ok1 := usingCortexMap[o.Id]
									//isEnable, ok1 := enableCortexMap[enableKey]

									//if !(ok1 && sourceId == id && len(originCortexTask.TaskId) <= 0) && !(originCortexTask.IsEnable == 1 && len(originCortexTask.TaskId) <= 0) {
									//	continue
									//}

									//e.Log.Trace("你好吗吗吗吗吗吗吗吗吗吗:", enableKey, enableCortexMap, isEnable, ok1)
									if len(originCortexTask.TaskId) > 0 {
										continue
									}
									task.Id = originCortexTask.Id
								}

								if pushOk {
									task.PushUrl = pushInfo.PushUrl
								}
								if pullOk {
									task.CustomOutUrl = pullInfo.PullUrl
								}

								task.Id = originCortexTask.Id
								task.CreateTime = now
								task.UpdateTime = now
								task.TaskState = 3
								isEnable := enableCortexMap[enableKey]
								e.Log.Trace("存储配置的输出点：", enableKey, enableCortexMap[enableKey])
								task.IsEnable = isEnable //是否配置 1是 2否
								count++
								addCortexItems = append(addCortexItems, task)
							}
						} else if v.MediaProcess.MediaType == 3 {
							if off.Area == 3 && ((off.StreamType == 2 && off.TrafficOffType == 1) || off.TrafficOffType == 2 || off.TrafficOffType == 6) {
								key1 := key + models.Delimiter + "1"
								inputAuditTaskKeys[key1] = struct{}{}
								key2 := key + models.Delimiter + "2"
								inputAuditTaskKeys[key2] = struct{}{}
							} else {
								key3 := key + models.Delimiter + "0"
								inputAuditTaskKeys[key3] = struct{}{}
							}

							//如果不存在就加入添加列表
							originTask, ok := originAuditKeys[key]
							if ok && len(originTask.TaskId) > 0 {
								continue
							}
							task := live_schedule.AuditTask{
								TenantId:     c.TenantId,
								ChannelId:    c.ChannelId,
								SourceId:     id,
								TrafficOnId:  up,
								MediaId:      v.TransferMediaPointId,
								TrafficOffId: down,
								PushUrl:      originTask.PushUrl,
								CustomOutUrl: originTask.CustomOutUrl,
							}
							if pushOk {
								task.PushUrl = pushInfo.PushUrl
							}
							if pullOk {
								task.CustomOutUrl = pullInfo.PullUrl
							}
							task.Id = originTask.Id
							task.CreateTime = now
							task.UpdateTime = now
							workTotalNum++
							addAuditItems = append(addAuditItems, task)
						} else if v.MediaProcess.MediaType == 4 {
							if off.Area == 3 && ((off.StreamType == 2 && off.TrafficOffType == 1) || off.TrafficOffType == 2 || off.TrafficOffType == 6) {
								key1 := key + models.Delimiter + "1"
								inputTransCodeTaskKeys[key1] = struct{}{}
								key2 := key + models.Delimiter + "2"
								inputTransCodeTaskKeys[key2] = struct{}{}
							} else {
								key3 := key + models.Delimiter + "0"
								inputTransCodeTaskKeys[key3] = struct{}{}
							}

							//如果不存在就加入添加列表
							originTask, ok := originTransCodeKeys[key]
							if ok && len(originTask.TaskId) > 0 {
								continue
							}
							task := live_schedule.ChannelTransCodeTask{
								TenantId:     c.TenantId,
								ChannelId:    c.ChannelId,
								SourceId:     id,
								TrafficOnId:  up,
								MediaId:      v.TransferMediaPointId,
								TrafficOffId: down,
								PushUrl:      originTask.PushUrl,
								CustomOutUrl: originTask.CustomOutUrl,
							}
							if pushOk {
								task.PushUrl = pushInfo.PushUrl
							}
							if pullOk {
								task.CustomOutUrl = pullInfo.PullUrl
							}
							task.Id = originTask.Id
							task.CreateTime = now
							task.UpdateTime = now
							workTotalNum++
							count++
							addTransCodeItems = append(addTransCodeItems, task)
						} else if v.MediaProcess.MediaType == 5 {
							key := strconv.Itoa(id) + models.Delimiter + strconv.Itoa(up) + models.Delimiter + strconv.Itoa(v.TransferMediaPointId)
							inputRecordTaskKeys[key] = struct{}{}
							//如果不存在就加入添加列表
							originTask, ok := originRecordKeys[key]
							if ok && len(originTask.RecordTaskId) > 0 {
								continue
							}
							task := live_schedule.SourceRecordTask{
								TenantId:    c.TenantId,
								ChannelId:   c.ChannelId,
								SourceId:    id,
								TrafficOnId: up,
								MediaId:     v.TransferMediaPointId,
								PushUrl:     originTask.PushUrl,
							}
							task.Id = originTask.Id
							task.CreateTime = now
							task.UpdateTime = now
							workTotalNum++
							addRecordItems = append(addRecordItems, task)
						} else if v.MediaProcess.MediaType == 6 {
							if off.Area == 3 && ((off.StreamType == 2 && off.TrafficOffType == 1) || off.TrafficOffType == 2 || off.TrafficOffType == 6) {
								key1 := key + models.Delimiter + "1"
								inputWatermarkTaskKeys[key1] = struct{}{}
								key2 := key + models.Delimiter + "2"
								inputWatermarkTaskKeys[key2] = struct{}{}
							} else {
								key3 := key + models.Delimiter + "0"
								inputWatermarkTaskKeys[key3] = struct{}{}
							}

							//如果不存在就加入添加列表
							originTask, ok := originWatermarkKeys[key]
							if ok && len(originTask.TaskId) > 0 {
								continue
							}
							task := live_schedule.WaterMarkTransTask{
								TenantId:     c.TenantId,
								ChannelId:    c.ChannelId,
								SourceId:     id,
								TrafficOnId:  up,
								MediaId:      v.TransferMediaPointId,
								TrafficOffId: down,
								PushUrl:      originTask.PushUrl,
								CustomOutUrl: originTask.CustomOutUrl,
							}
							if pushOk {
								task.PushUrl = pushInfo.PushUrl
							}
							if pullOk {
								task.CustomOutUrl = pullInfo.PullUrl
							}
							task.Id = originTask.Id
							task.CreateTime = now
							task.UpdateTime = now
							workTotalNum++
							count++
							addWatermarkItems = append(addWatermarkItems, task)
						}
					}
				}
			}
		}

		//存储需要删除的信源id
		delSourceMap := make(map[int]struct{})
		//过滤出需要删除的节点
		for key, v := range originAllKeys {
			v := v
			_, ok := inputTaskKeys[key]
			if !ok {
				delSourceMap[v.SourceId] = struct{}{}
				delConvertItems = append(delConvertItems, v)
				workTotalNum++
				count--
			}
		}
		e.Log.Trace("所有的key", originCortexAllKeys)
		e.Log.Trace("输入的所有的key", inputCortexTaskKeys)
		for key, v := range originCortexAllKeys {
			v := v
			_, ok := inputCortexTaskKeys[key]
			if !ok {
				delSourceMap[v.SourceId] = struct{}{}
				delCortexItems = append(delCortexItems, v)
				workTotalNum++
				count--
			}
		}
		for key, v := range originRecordKeys {
			v := v
			_, ok := inputRecordTaskKeys[key]
			if !ok {
				delRecordItems = append(delRecordItems, v)
				workTotalNum++
			}
		}
		//过滤出需要删除的审核任务节点
		for key, v := range originAuditKeysAllKeys {
			v := v
			_, ok := inputAuditTaskKeys[key]
			if !ok {
				delAuditItems = append(delAuditItems, v)
				workTotalNum++
				count--
			}
		}

		//过滤出需要删除的转码任务节点
		for key, v := range originTransCodeAllKeys {
			v := v
			_, ok := inputTransCodeTaskKeys[key]
			if !ok {
				delTransCodeItems = append(delTransCodeItems, v)
				workTotalNum++
				count--
			}
		}

		e.Log.Trace("需要删除的转码任务:", delTransCodeItems)
		//过滤出需要删除的水印任务节点
		for key, v := range originWatermarkAllKeys {
			v := v
			_, ok := inputWatermarkTaskKeys[key]
			if !ok {
				delWatermarkItems = append(delWatermarkItems, v)
				workTotalNum++
				count--
			}
		}

		//获取分发限制数
		limitRes, err := streamHubService.DispatchLimit(&StreamHubDto.StreamHubReq{TenantId: c.TenantId})
		if err != nil {
			e.Log.Errorf("DispatchLimit error:%s ", err)
		}
		if limitRes != nil && limitRes.Count > 0 && count > limitRes.Rows[0].Quota.MediaScheduler.ParanDispatchNum {
			param1 := make(map[string]interface{})
			param1["channel_state"] = basic.ChannelState
			param1["progress"] = "0"
			param1["update_time"] = now
			err = e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", c.ChannelId).Updates(&param1).Error
			if err != nil {
				e.Log.Errorf("update ChannelBasic error:%s \r\n", err)
				return err
			}
			return errors.New(string(apis.ScheduleCountError.Code))
			e.Log.Trace("使用调度数情况:", count, limitRes.Rows[0].Quota.MediaScheduler.ParanDispatchNum)
		}

		//获取所有的信源
		sources := make([]live_schedule.SourceBasic, 0, 5)
		err = e.Orm.Model(live_schedule.SourceBasic{}).Where(" id in ?", sourcesIds).Find(&sources).Error
		if err != nil {
			e.Log.Errorf("SaveWorkflow get SourceBasic error:%s \r\n", err)
			return err
		}
		for _, v := range sources {
			v := v
			sourcesMap[v.Id] = v
		}
		//获取信源详情
		details := make([]live_schedule.SourceDetail, 0, 5)
		err = e.Orm.Model(live_schedule.SourceDetail{}).Where("tenant_id=? and source_id in ?", basic.TenantId, sourcesIds).Find(&details).Error
		if err != nil {
			e.Log.Errorf("SaveWorkflow get SourceDetail error:%s \r\n", err)
			return err
		}

		detailMap := make(map[int][]live_schedule.SourceDetail)
		for _, v := range details {
			v := v
			detail, ok := detailMap[v.SourceId]
			if !ok {
				detail = make([]live_schedule.SourceDetail, 0, 5)
			}
			detail = append(detail, v)
			detailMap[v.SourceId] = detail
		}

		//获取媒体处理
		mediaProcess := make([]live_schedule.MediaProcess, 0, 5)
		err = e.Orm.Model(live_schedule.MediaProcess{}).Where("id in ?", mediaIds).Find(&mediaProcess).Error
		if err != nil {
			e.Log.Errorf("get MediaProcess error:%s \r\n", err)
			return err
		}
		mediaProcessMap := make(map[int]live_schedule.MediaProcess)
		for _, m := range mediaProcess {
			m := m
			mediaProcessMap[m.MediaType] = m
		}

		upMediaIds := make([]int, 0, 5)
		//处理转码软矩阵无缝切换
		for _, m := range c.WorkflowMedia {
			if m.MediaProcess.MediaType != 2 || m.MediaProcess.EnableSwitch == 2 {
				continue
			}
			err = e.Orm.Model(live_schedule.UpMedia{}).Select("up_id").Where("channel_id = ? and media_id = ?", c.ChannelId, m.MediaProcess.Id).Find(&upMediaIds).Error
			if err != nil {
				e.Log.Errorf("get UpMedia error:%s \r\n", err)
				return err
			}

			//如果矩阵原上车点和现在上车点不一致需要更新切换任务
			if !reflect.DeepEqual(m.TransferUpPointId, upMediaIds) && ext.ExtConfig.ServiceConfig.EnableSwitch == 1 {
				workTotalNum = workTotalNum + len(cortexOutputs)
				cortexTaskService.UpdateCortexSwitchTask(workTotalNum, &partNum, cortexTasks, addCortexItems, cortexOutputs, m.TransferUpPointId, upMap, sourcesMap, basic, streamHubService, liveBasicService)
			}
		}
		//删除节点
		if len(delConvertItems) > 0 {
			e.Log.Trace("删除协转任务")
			err = convertTaskService.DeleteChannelTask(&wg, workTotalNum, &partNum, delConvertItems, sourcesMap, basic, liveService, streamHubService, liveBasicService)
			if err != nil {
				e.Log.Trace("DeleteChannelTask error:", err)
				return err
			}
		}

		//删除棋盘格任务
		if len(delCortexItems) > 0 {
			e.Log.Trace("删除棋盘格任务")
			err = cortexTaskService.DeleteCortexTask(&wg, workTotalNum, &partNum, delCortexItems, sourcesMap, basic, liveService, streamHubService, liveBasicService)
			if err != nil {
				e.Log.Errorf("DeleteCortexTask error:%s \r\n", err)
				return err
			}
		}

		//删除审核任务
		if len(delAuditItems) > 0 {
			e.Log.Trace("删除审核任务")
			err = auditTaskService.DeleteAuditTask(&wg, workTotalNum, &partNum, delAuditItems, sourcesMap, basic, liveService, streamHubService, liveBasicService)
			if err != nil {
				e.Log.Errorf("DeleteAuditTask error:%s \r\n", err)
				return err
			}
		}

		//删除录制任务
		if len(delRecordItems) > 0 {
			e.Log.Trace("删除录制任务")
			err = recordTaskService.DeleteRecordTask(&wg, workTotalNum, &partNum, delRecordItems, sourcesMap, basic, liveService, streamHubService, liveBasicService)
			if err != nil {
				e.Log.Errorf("DeleteRecordTask error:%s \r\n", err)
				return err
			}
		}

		//删除转码任务
		if len(delTransCodeItems) > 0 {
			err = transCodeTaskService.DeleteTransCodeTask(&wg, workTotalNum, &partNum, delTransCodeItems, sourcesMap, basic, liveService, streamHubService, liveBasicService)
			if err != nil {
				e.Log.Errorf("DeleteTransCodeTask error:%s \r\n", err)
				return err
			}
		}

		//删除水印任务
		if len(delWatermarkItems) > 0 {
			err = waterMarkTaskService.DeleteWaterMarkTransTask(&wg, workTotalNum, &partNum, delWatermarkItems, sourcesMap, basic, liveService, streamHubService, liveBasicService)
			if err != nil {
				e.Log.Errorf("DeleteWaterMarkTransTask error:%s \r\n", err)
				return err
			}
		}

		e.Log.Trace("需要添加的协转任务：", workTotalNum, addConvertItems, addCortexItems, addAuditItems, addRecordItems, addTransCodeItems)

		//添加协转任务
		if len(addConvertItems) > 0 {
			err = convertTaskService.CreateChannelTask(&wg, workTotalNum, &partNum, addConvertItems, upMap, offMap, basic, mediaProcessMap[1], sourcesMap)
			if err != nil {
				e.Log.Errorf("CreateChannelTask error:%s \r\n", err)
				return err
			}
		}
		//新增软矩阵任务
		if len(addCortexItems) > 0 {
			err = e.Orm.Save(&addCortexItems).Error
			if err != nil {
				e.Log.Errorf("save CortexTask error:%s \r\n", err)
				//tx.Rollback()
				return err
			}

			//存储软矩阵任务
			//taskMap := make(map[string]live_schedule.CortexTask)
			needCreateCortexTasks := make([]live_schedule.CortexTask, 0, 5)
			for _, v := range addCortexItems {
				v := v
				//key := strconv.Itoa(v.TrafficOnId) + models.Delimiter + strconv.Itoa(v.SourceId) + models.Delimiter + strconv.Itoa(v.CortexOutputId) + models.Delimiter + strconv.Itoa(v.TrafficOffId)
				//tasks := useCortexTaskMap[key]
				//if len(tasks) > 0 {
				//	off := offMap[v.TrafficOffId]
				//	if off.StreamType == 1 {
				//		continue
				//	}
				//	//如果需要添加的任务存在相同输出名但是信源不同的输出任务则跳过
				//	sourceId := usingCortexMap[v.CortexOutputId]
				//	if sourceId != v.SourceId {
				//		continue
				//	}
				//	//存储软矩阵任务
				//	originKey := v.TenantId + models.Delimiter + strconv.Itoa(v.ChannelId) + models.Delimiter + strconv.Itoa(v.TrafficOnId) + models.Delimiter + strconv.Itoa(v.MediaProcessId) + models.Delimiter + strconv.Itoa(v.CortexOutputId) + models.Delimiter + strconv.Itoa(v.TrafficOffId)
				//	taskMap[originKey] = v
				//	e.Log.Trace("所有任务：", taskMap)
				//}
				//针对添加的新的上车点
				sourceId, ok := usingCortexMap[v.CortexOutputId]
				e.Log.Trace("需要添加的任务", usingCortexMap, v.CortexOutputId, sourceId)
				if (!ok || sourceId != v.SourceId) && !(v.IsEnable == 1 && v.PushUrl != "") {
					continue
				}
				needCreateCortexTasks = append(needCreateCortexTasks, v)
			}

			//for k, off := range offMap {
			//	//如果是srt拉流或者httpts走的还是内网
			//	if off.TrafficOffType == 3 || (off.TrafficOffType == 4 && off.StreamType == 2) {
			//		off.Area = 2
			//		offMap[k] = off
			//	}
			//	for _, up := range upMap {
			//		for _, output := range outputMap {
			//			originKey := basic.TenantId + models.Delimiter + strconv.Itoa(basic.Id) + models.Delimiter + strconv.Itoa(up.Id) + models.Delimiter + strconv.Itoa(mediaProcessMap[2].Id) + models.Delimiter + strconv.Itoa(output.Id) + models.Delimiter + strconv.Itoa(off.Id)
			//			e.Log.Trace("匹配到的任务：", originKey)
			//			task, _ := taskMap[originKey]
			//			_, ok := sourcesMap[up.SourceId]
			//			if !ok {
			//				return errors.New("信源已被删除")
			//			}
			//			e.Log.Trace("匹配到的任务：", originKey)
			//			task, ok1 := taskMap[originKey]
			//			if !ok1 {
			//				e.Log.Trace("未找到任务：", originKey)
			//				continue
			//			}
			//
			//			//已经有的输出节点跳过
			//			if len(task.TaskId) > 0 {
			//				continue
			//			}
			//			//if len(task.PushUrl) <= 0 && off.StreamType == 1 || (off.IsCustom == 1 && len(task.CustomOutUrl) <= 0) && flag == 1{
			//			//	return errors.New("有未填写的输出地址")
			//			//}
			//			workTotalNum++
			//			needCreateCortexTasks = append(needCreateCortexTasks, task)
			//		}
			//	}
			//}

			e.Log.Trace("需要添加的任务", needCreateCortexTasks)
			err = cortexTaskService.CreateCortexTasks(&wg, workTotalNum, &partNum, needCreateCortexTasks, offMap, upMap, basic, sourcesMap, cortexOutputMap)
			if err != nil {
				e.Log.Errorf("CreateCortexTasks error:%s \r\n", err)
				//tx.Rollback()
				return err
			}

		}
		//新增审核任务
		if len(addAuditItems) > 0 {
			//获取审核任务
			audits := make([]live_schedule.AuditTask, 0, 5)
			auditMap := make(map[int]live_schedule.AuditTask)
			err = e.Orm.Model(&live_schedule.AuditTask{}).Where("source_id in ? and audit_task_id!=''", sourcesIds).Find(&audits).Error
			if err != nil {
				e.Log.Errorf("get AuditTask error:%s \r\n", err)
				return err
			}
			for _, v := range audits {
				v := v
				auditMap[v.SourceId] = v
			}
			e.Log.Trace("需要添加的审核任务", addAuditItems)
			err = auditTaskService.SaveAuditTask(&wg, workTotalNum, &partNum, addAuditItems, upMap, offMap, basic, mediaProcessMap[3], sourcesMap, auditMap)
			if err != nil {
				e.Log.Errorf("SaveAuditTask error:%s \r\n", err)
				return err
			}
		}
		//新增录制任务
		if len(addRecordItems) > 0 {
			err = recordTaskService.ProcessSourceRecordTask(&wg, workTotalNum, &partNum, addRecordItems, upMap, basic, mediaProcessMap[5], sourcesMap)
			if err != nil {
				e.Log.Errorf("ProcessSourceRecordTask error:%s \r\n", err)
				return err
			}
		}
		//新增转码任务
		if len(addTransCodeItems) > 0 {
			err = transCodeTaskService.CreateChannelTransCodeTask(&wg, workTotalNum, &partNum, addTransCodeItems, upMap, offMap, basic, mediaProcessMap[4], sourcesMap)
			if err != nil {
				e.Log.Errorf("CreateChannelTransCodeTask error:%s \r\n", err)
				return err
			}
		}
		//新增水印任务
		if len(addWatermarkItems) > 0 {
			err = waterMarkTaskService.CreateWatermarkTasks(&wg, workTotalNum, &partNum, addWatermarkItems, upMap, offMap, basic, mediaProcessMap[6], sourcesMap)
			if err != nil {
				e.Log.Errorf("CreateWatermarkTasks error:%s \r\n", err)
				return err
			}
		}
	}

	//上车点媒体关系
	var upMedia live_schedule.UpMedia
	err = e.Orm.Model(live_schedule.UpMedia{}).Where("tenant_id=? and channel_id=? and media_id in ?", c.TenantId, c.ChannelId, mediaIds).Delete(&upMedia).Error
	if err != nil {
		e.Log.Errorf("ChannelTaskService UpMedia_delete error:%s \r\n", err)
		return err
	}
	upMedias := make([]live_schedule.UpMedia, 0, 5)
	//下车点媒体关系
	var mediaOff live_schedule.MediaOff
	err = e.Orm.Model(live_schedule.MediaOff{}).Where("tenant_id=? and channel_id=? and media_id in ?", c.TenantId, c.ChannelId, mediaIds).Delete(&mediaOff).Error
	if err != nil {
		e.Log.Errorf("ChannelTaskService MediaOff_delete error:%s \r\n", err)
		return err
	}
	mediaOffs := make([]live_schedule.MediaOff, 0, 5)
	for _, v := range c.WorkflowMedia {
		mediaId := v.MediaProcess.Id

		for _, up := range v.TransferUpPointId {
			upMedia := live_schedule.UpMedia{}
			upMedia.TenantId = c.TenantId
			upMedia.ChannelId = c.ChannelId
			upMedia.MediaId = mediaId
			upMedia.UpId = up
			upMedia.CreateTime = now
			upMedia.UpdateTime = now
			upMedias = append(upMedias, upMedia)
		}
		for _, down := range v.TransferDownPointId {
			mediaOff := live_schedule.MediaOff{}
			mediaOff.TenantId = c.TenantId
			mediaOff.ChannelId = c.ChannelId
			mediaOff.MediaId = mediaId
			mediaOff.OffId = down
			mediaOff.CreateTime = now
			mediaOff.UpdateTime = now
			mediaOffs = append(mediaOffs, mediaOff)
		}

	}

	if len(upMedias) > 0 {
		err = e.Orm.Model(live_schedule.UpMedia{}).Create(upMedias).Error
		if err != nil {
			e.Log.Errorf("ChannelTaskService UpMedia_create error:%s \r\n", err)
			return err
		}
	}

	if len(mediaOffs) > 0 {
		err = e.Orm.Model(live_schedule.MediaOff{}).Create(mediaOffs).Error
		if err != nil {
			e.Log.Errorf("ChannelTaskService MediaOff_create error:%s \r\n", err)
			return err
		}
	}

	wg.Wait()
	//if workTotalNum == 0 {
	//	err = e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", c.ChannelId).Update("channel_state", basic.ChannelState).Error
	//	if err != nil {
	//		e.Log.Errorf("update ChannelBasic error:%s \r\n", err)
	//		return err
	//	}
	//}

	return nil
}

func (e *ChannelTaskService) GetWorkflow(c *dto.GetWorkflowReq) (dto.GetWorkflowRes, error) {

	pushInfoMap := make(map[string][]dto.PushInfo)
	pullInfoMap := make(map[string][]dto.PullInfo)
	res := dto.GetWorkflowRes{}
	//获取上车点
	ups := make([]live_schedule.TrafficOnModel, 0, 5)
	err := e.Orm.Model(live_schedule.TrafficOn{}).
		Select("traffic_on.*,source_basic.protocol_type").
		Joins("left join source_basic on source_basic.id = traffic_on.source_id").
		Where("traffic_on.tenant_id=? and traffic_on.channel_id=?", c.TenantId, c.ChannelId).
		Find(&ups).
		Error
	if err != nil {
		e.Log.Errorf("Service GetWorkflow getTrafficOn error:%s \r\n", err)
		return res, err
	}

	workflowUps := make([]dto.WorkflowUp, 0, 5)
	groupUpMap := make(map[string]live_schedule.TrafficOnModel)
	//上车点和信源组id映射
	upGroupMap := make(map[int]string)
	upMap := make(map[int]live_schedule.TrafficOnModel)
	upSourceMap := make(map[int]int)
	sourceIds := make([]int, 0, 5)
	for _, v := range ups {
		v := v
		if v.IsGroup == 1 {
			groupUpMap[strconv.Itoa(v.Id)] = v
		} else {
			sourceIds = append(sourceIds, v.SourceId)
			upSourceMap[v.SourceId] = v.Id
		}
		upMap[v.Id] = v
	}

	//获取信源基本详情
	upSourceInfoMap := make(map[int]*dto.SourceModelRes)
	basicService := &SourceBasicService{}
	basicService.Orm = e.Orm
	basicService.Log = e.Log
	var count int64
	pagination := commonDto.Pagination{PageIndex: 1, PageSize: 1000}
	//信源状态映射
	sourceStateMap := make(map[int]int)
	//获取原始流信息
	isAllRed := true
	sourceMap := make(map[int]live_schedule.SourceBasic)
	isExistUdp := false
	transMap := make(map[int]live_schedule.SourceTemplate)
	if len(sourceIds) > 0 {
		basics, err := basicService.GetPage(&dto.SourceBasicGetPageReq{Pagination: pagination, SourceIds: sourceIds, IsDelete: 2, Kind: 5}, &count)
		if err != nil {
			e.Log.Errorf("ChannelTaskService GetSourceBasic error:%s \r\n", err)
			return res, err
		}
		for _, v := range *basics {
			v := v
			sourceMap[v.Basic.Id] = *v.Basic
			if v.Basic.Kind == 4 && v.Basic.SourceState == 3 {
				v.Basic.SourceState = 1
			}
			sourceStateMap[v.Basic.Id] = v.Basic.SourceState
			e.Log.Trace("信源状v态", v.Basic.SourceName, v.Basic.SourceState)
			if v.Basic.SourceState == 1 {
				isAllRed = false
			}
			id, ok := upSourceMap[v.Basic.Id]
			if ok {
				upSourceInfoMap[id] = &v
			}
			if !isExistUdp && v.Basic.ProtocolType == 5 {
				isExistUdp = true
			}
			for _, t := range v.TransCodeTasks {
				t := t
				transMap[t.SourceTemplate.Id] = t.SourceTemplate
			}
		}
	}

	//上车点
	for _, v := range ups {
		v := v
		if v.IsGroup == 1 {
			continue
		}

		source, ok := sourceMap[v.SourceId]
		v.SourceArea = source.Area
		v.SourceStreamType = source.StreamType
		v.SourceType = source.SourceType
		if v.GroupId == "0" {
			v.GroupId = ""
		}

		groupUp, ok := groupUpMap[v.GroupId]
		workflowUp := dto.WorkflowUp{}
		workflowUp.TrafficOn = &v
		if ok {
			workflowUp.Id = strconv.Itoa(groupUp.Id)
			workflowUp.GroupTrafficOn = &groupUp
			workflowUp.TrafficOn.GroupId = strconv.Itoa(groupUp.Id)
		}
		upGroupMap[v.Id] = v.GroupId
		state := sourceStateMap[v.SourceId]

		//如果信源异常或者 上车点被删除（ProtocolType == 0）
		if state == 3 && source.ProtocolType != 5 || v.ProtocolType == 0 {
			workflowUp.Status = "2"
		} else {
			workflowUp.Status = "1"
		}

		upSourceInfo, ok1 := upSourceInfoMap[v.Id]
		if ok1 {
			workflowUp.SourceModelRes = upSourceInfo
		} else {
			//处理上车点为相同信源的情况
			upId, ok2 := upSourceMap[v.SourceId]
			if ok2 {
				workflowUp.SourceModelRes = upSourceInfoMap[upId]
			}
		}

		//如果转码任务被删除 设置为0
		if v.StreamType == 2 {
			_, ok2 := transMap[v.TransRecordId]
			if !ok2 {
				workflowUp.TrafficOn.ProtocolType = 0
			}
		}
		if v.TrafficOnState == 2 {
			workflowUp.Status = "0"
		}
		workflowUps = append(workflowUps, workflowUp)
	}
	res.WorkflowUp = &workflowUps

	//获取下车点信息
	offs := make([]live_schedule.TrafficOff, 0, 5)
	err = e.Orm.Model(live_schedule.TrafficOff{}).Where("tenant_id=? and channel_id=?", c.TenantId, c.ChannelId).Find(&offs).Error
	if err != nil {
		e.Log.Errorf("Service GetWorkflow getTrafficOff error:%s \r\n", err)
		return res, err
	}
	offIds := make([]int, 0, 5)
	offMap := make(map[int]live_schedule.TrafficOff)
	for _, v := range offs {
		v := v
		offIds = append(offIds, v.Id)
		offMap[v.Id] = v
	}

	//获取任务
	offTasks := make([]live_schedule.ChannelTask, 0, 5)
	err = e.Orm.Model(live_schedule.ChannelTask{}).Where("traffic_off_id in ?", offIds).Find(&offTasks).Error
	if err != nil {
		e.Log.Errorf(" ChannelTask error:%s \r\n", err)
		return res, err
	}
	//获取媒体处理信息
	mediaIds := make([]int, 0, 5)
	mediaAllIds := make([]int, 0, 5)
	medias := make([]live_schedule.MediaProcess, 0, 5)
	e.Orm.Model(live_schedule.MediaProcess{}).Where("tenant_id=? and channel_id=?", c.TenantId, c.ChannelId).Find(&medias)

	for _, v := range medias {
		mediaIds = append(mediaIds, v.Id)
		mediaAllIds = append(mediaAllIds, v.Id)
	}

	//获取上车点和媒体处理关系
	upMedias := make([]live_schedule.UpMedia, 0, 5)
	err = e.Orm.Model(live_schedule.UpMedia{}).Where("tenant_id=? and channel_id=? and media_id in ? ", c.TenantId, c.ChannelId, mediaAllIds).Find(&upMedias).Error
	if err != nil {
		e.Log.Errorf("Service GetWorkflow getUpMedia error:%s \r\n", err)
		return res, err
	}
	upMediaMap := make(map[int][]int)
	for _, v := range upMedias {
		upMedia, ok := upMediaMap[v.MediaId]
		if !ok {
			upMedia = make([]int, 0, 5)
		}
		upMedia = append(upMedia, v.UpId)
		upMediaMap[v.MediaId] = upMedia
	}

	//获取下车点和媒体处理关系
	offMedias := make([]live_schedule.MediaOff, 0, 5)
	err = e.Orm.Model(live_schedule.MediaOff{}).Where("tenant_id=? and channel_id=? and media_id in ? ", c.TenantId, c.ChannelId, mediaIds).Find(&offMedias).Error
	if err != nil {
		e.Log.Errorf("Service GetWorkflow getMediaOff error:%s \r\n", err)
		return res, err
	}
	offMediaMap := make(map[int][]int)
	for _, v := range offMedias {
		offMedia, ok := offMediaMap[v.MediaId]
		if !ok {
			offMedia = make([]int, 0, 5)
		}
		offMedia = append(offMedia, v.OffId)
		offMediaMap[v.MediaId] = offMedia
	}

	workflowMedias := make([]dto.WorkflowMedia, 0, 5)
	for _, v := range medias {
		v := v
		workflowMedia := dto.WorkflowMedia{}
		workflowMedia.TransferMediaPointId = v.Id
		workflowMedia.TransferUpPointId = upMediaMap[v.Id]
		//审核处理没有下车点
		workflowMedia.TransferDownPointId = offMediaMap[v.Id]

		groupId := ""
		for _, t := range upMediaMap[v.Id] {
			id := upGroupMap[t]
			if len(id) > 0 {
				num, _ := strconv.Atoi(id)
				if num > 0 {
					groupId = id
				}
				break
			}
		}
		//state := 3
		//states := make([]int, 0, 5)
		//switch v.MediaType {
		//case 1:
		//	err = e.Orm.Model(&live_schedule.ChannelTask{}).Select("task_state").Where("media_id = ?", v.Id).Find(&states).Error
		//	if err != nil {
		//		e.Log.Errorf(" get ChannelTask error:%s \r\n", err)
		//
		//	}
		//case 2:
		//	err = e.Orm.Model(&live_schedule.CortexTask{}).Select("task_state").Where("media_id = ? and task_state!=''", v.Id).Find(&states).Error
		//	if err != nil {
		//		e.Log.Errorf(" get CortexTask error:%s \r\n", err)
		//
		//	}
		//	fmt.Println("kkkkkkkk:", states)
		//case 3:
		//	err = e.Orm.Model(&live_schedule.AuditTask{}).Select("task_state").Where("media_id = ?", v.Id).Find(&states).Error
		//	if err != nil {
		//		e.Log.Errorf(" get AuditTask error:%s \r\n", err)
		//
		//	}
		//case 4:
		//	err = e.Orm.Model(&live_schedule.ChannelTransCodeTask{}).Select("task_state").Where("media_id = ?", v.Id).Find(&states).Error
		//	if err != nil {
		//		e.Log.Errorf(" get ChannelTransCodeTask error:%s \r\n", err)
		//
		//	}
		//case 5:
		//	err = e.Orm.Model(&live_schedule.SourceRecordTask{}).Select("task_state").Where("media_id = ?", v.Id).Find(&states).Error
		//	if err != nil {
		//		e.Log.Errorf(" get SourceRecordTask error:%s \r\n", err)
		//
		//	}
		//case 6:
		//	err = e.Orm.Model(&live_schedule.WaterMarkTransTask{}).Select("task_state").Where("media_id = ?", v.Id).Find(&states).Error
		//	if err != nil {
		//		e.Log.Errorf(" get WaterMarkTransTask error:%s \r\n", err)
		//
		//	}
		//}

		//if allSame(states, 1) {
		//	state = 1
		//} else if allSame(states, 2) {
		//	state = 2
		//} else if allSame(states, 3) {
		//	state = 3
		//} else {
		//	state = 4
		//}
		//v.MediaState = state
		workflowMedia.GroupId = groupId
		if workflowMedia.GroupId == "0" {
			workflowMedia.GroupId = ""
		}
		workflowMedia.MediaProcess = v
		if isAllRed && !isExistUdp {
			workflowMedia.Status = "2"
		} else {
			workflowMedia.Status = "1"
		}
		if workflowMedia.MediaProcess.MediaState == 2 {
			workflowMedia.Status = "0"
		}

		if v.MediaType == 3 {
			results := make([]live_schedule.AuditResult, 0, 5)
			err = e.Orm.Model(&live_schedule.AuditResult{}).Where("source_id in ?", sourceIds).Find(&results).Error
			if err != nil {
				e.Log.Errorf("get AuditResult error:%s \r\n", err)
				return res, err
			}
			statics := dto.AuditStatisticRes{}
			type1 := 0
			type2 := 0
			type3 := 0
			type4 := 0
			type5 := 0
			type6 := 0
			for _, v := range results {
				switch v.ViolationType {
				case 1:
					workflowMedia.Status = 3
					type1++
				case 2:
					workflowMedia.Status = 3
					type2++
				case 3:
					workflowMedia.Status = 3
					type3++
				case 4:
					workflowMedia.Status = 3
					type4++
				case 5:
					workflowMedia.Status = 3
					type5++
				default:
					type6++
				}

			}
			statics.Type1Num = type1
			statics.Type2Num = type2
			statics.Type3Num = type3
			statics.Type4Num = type4
			statics.Type5Num = type5
			statics.Type6Num = type6
			workflowMedia.AuditStatisticRes = statics
		}

		if v.MediaType == 2 {
			outputs := make([]live_schedule.CortexOutput, 0, 5)
			err = e.Orm.Model(&live_schedule.CortexOutput{}).Where("media_process_id=?", v.Id).Find(&outputs).Error
			if err != nil {
				e.Log.Errorf("get CortexOutput error:%s \r\n", err)
			}
			outMap := make(map[int]live_schedule.CortexOutput)
			for _, output := range outputs {
				outMap[output.Id] = output
			}
			offs := offMediaMap[v.Id]
			for _, offId := range offs {
				off, ok := offMap[offId]
				if off.StreamType == 1 {
					pushInfos, ok := pushInfoMap[strconv.Itoa(v.Id)]
					if !ok {
						pushInfos = make([]dto.PushInfo, 0, 5)
					}
					pushInfo := dto.PushInfo{}
					if len(off.PushInfos) > 0 {
						err = json.Unmarshal([]byte(off.PushInfos), &pushInfos)
						if err != nil {
							e.Log.Errorf("Unmarshal PushInfos error:%s \r\n", err)
						}
					}

					for _, out := range outputs {
						isExisted := false
						for _, info := range pushInfos {
							if info.CortexOutName == out.OutName {
								isExisted = true
								break
							}
						}
						if !isExisted {
							pushInfo.CortexOutId = out.Id
							pushInfo.MediaId = out.MediaProcessId
							pushInfo.OffId = offId
							pushInfo.MediaId = v.Id
							pushInfo.PushUrl = ""
							pushInfo.CortexOutName = out.OutName
							pushInfos = append(pushInfos, pushInfo)
						}

					}

					pushInfoMap[strconv.Itoa(offId)] = pushInfos
				}
				if ok && off.StreamType == 2 && off.TrafficOffType == 2 {
					pullInfos, ok := pullInfoMap[strconv.Itoa(v.Id)]
					if !ok {
						pullInfos = make([]dto.PullInfo, 0, 5)
					}
					pullInfo := dto.PullInfo{}
					if len(off.PullInfos) > 0 {
						err = json.Unmarshal([]byte(off.PullInfos), &pullInfos)
						if err != nil {
							e.Log.Errorf("Unmarshal PullInfos error:%s \r\n", err)
						}
					}
					for _, out := range outputs {
						isExisted := false
						for _, info := range pullInfos {
							if info.CortexOutName == out.OutName {
								isExisted = true
								break
							}
						}
						if !isExisted {
							pullInfo.CortexOutId = out.Id
							pullInfo.MediaId = out.MediaProcessId
							pullInfo.OffId = offId
							pullInfo.MediaId = v.Id
							pullInfo.PullUrl = ""
							pullInfo.CortexOutName = out.OutName
							pullInfos = append(pullInfos, pullInfo)
						}

					}
					pullInfoMap[strconv.Itoa(offId)] = pullInfos
				}
			}
		}
		workflowMedias = append(workflowMedias, workflowMedia)
	}
	res.WorkflowMedia = &workflowMedias

	//获取协转任务
	tasks := make([]live_schedule.ChannelTask, 0, 5)
	err = e.Orm.Model(live_schedule.ChannelTask{}).Where("tenant_id=? and channel_id=?", c.TenantId, c.ChannelId).Find(&tasks).Error
	if err != nil {
		e.Log.Errorf("Service GetWorkflow getChannelTask error:%s \r\n", err)
		return res, err
	}

	now := time.Now().Unix()
	offChannelTaskMap := make(map[int][]live_schedule.ChannelTask)
	for _, v := range offTasks {
		v := v
		source := sourceMap[v.SourceId]
		if source.ProtocolType == 5 {
			t, ok := offChannelTaskMap[v.TrafficOffId]
			if !ok {
				t = make([]live_schedule.ChannelTask, 0, 5)
			}
			t = append(t, v)
			offChannelTaskMap[v.TrafficOffId] = t
		}
	}
	offChannelStateMap := make(map[int]int)
	for k, v := range offChannelTaskMap {
		for _, t := range v {
			if t.TaskState == 1 && t.StateUpdateTime+10 < now {
				offChannelStateMap[k] = 1
			} else {
				offChannelStateMap[k] = 2
				break
			}
		}
	}

	//获取软矩阵任务
	cortexTasks := make([]live_schedule.CortexTask, 0, 5)
	err = e.Orm.Model(live_schedule.CortexTask{}).Where("tenant_id=? and channel_id=?", c.TenantId, c.ChannelId).Find(&cortexTasks).Error
	if err != nil {
		e.Log.Errorf("Service GetWorkflow getCortexTask error:%s \r\n", err)
		return res, err
	}
	offCortexTaskMap := make(map[int][]live_schedule.CortexTask)
	for _, v := range cortexTasks {
		v := v
		source := sourceMap[v.SourceId]
		if source.ProtocolType == 5 {
			t, ok := offCortexTaskMap[v.TrafficOffId]
			if !ok {
				t = make([]live_schedule.CortexTask, 0, 5)
			}
			t = append(t, v)
			offCortexTaskMap[v.TrafficOffId] = t
		}
	}

	offCortexStateMap := make(map[int]int)
	for k, v := range offCortexTaskMap {
		for _, t := range v {
			if t.TaskState == 2 {
				offCortexStateMap[k] = 2
				break
			} else {
				offCortexStateMap[k] = 1
			}
		}
	}
	//infos := make([]dto.PushInfo, 0, 5)

	for _, v := range tasks {
		off, ok := offMap[v.TrafficOffId]
		if off.StreamType == 1 {
			pushInfos, ok := pushInfoMap[strconv.Itoa(v.TrafficOffId)]
			if !ok {
				pushInfos = make([]dto.PushInfo, 0, 5)
			}
			pushInfo := dto.PushInfo{}
			pushInfo.UpId = v.TrafficOnId
			pushInfo.SourceId = v.SourceId
			pushInfo.OffId = v.TrafficOffId
			pushInfo.MediaId = v.MediaId
			pushInfo.PushUrl = v.PushUrl
			pushInfos = append(pushInfos, pushInfo)
			pushInfoMap[strconv.Itoa(v.TrafficOffId)] = pushInfos
		}
		if ok && off.StreamType == 2 && off.TrafficOffType == 2 {
			pullInfos, ok := pullInfoMap[strconv.Itoa(v.TrafficOffId)]
			if !ok {
				pullInfos = make([]dto.PullInfo, 0, 5)
			}
			pullInfo := dto.PullInfo{}
			pullInfo.UpId = v.TrafficOnId
			pullInfo.SourceId = v.SourceId
			pullInfo.OffId = v.TrafficOffId
			pullInfo.MediaId = v.MediaId
			pullInfo.PullUrl = v.CustomOutUrl
			pullInfo.OriginPullUrl = v.OutUrl
			pullInfos = append(pullInfos, pullInfo)
			pullInfoMap[strconv.Itoa(v.TrafficOffId)] = pullInfos
		}
	}

	//获取转码任务
	tranTasks := make([]live_schedule.ChannelTransCodeTask, 0, 5)
	err = e.Orm.Model(live_schedule.ChannelTransCodeTask{}).Where("tenant_id=? and channel_id=?", c.TenantId, c.ChannelId).Find(&tranTasks).Error
	if err != nil {
		e.Log.Errorf("Service GetWorkflow getChannelTask error:%s \r\n", err)
		return res, err
	}
	for _, v := range tranTasks {
		off, ok := offMap[v.TrafficOffId]
		if off.StreamType == 1 {
			pushInfos, ok := pushInfoMap[strconv.Itoa(v.TrafficOffId)]
			if !ok {
				pushInfos = make([]dto.PushInfo, 0, 5)
			}
			pushInfo := dto.PushInfo{}
			pushInfo.UpId = v.TrafficOnId
			pushInfo.SourceId = v.SourceId
			pushInfo.OffId = v.TrafficOffId
			pushInfo.MediaId = v.MediaId
			pushInfo.PushUrl = v.PushUrl
			pushInfos = append(pushInfos, pushInfo)
			pushInfoMap[strconv.Itoa(v.TrafficOffId)] = pushInfos
		}
		if ok && off.StreamType == 2 && off.TrafficOffType == 2 {
			pullInfos, ok := pullInfoMap[strconv.Itoa(v.TrafficOffId)]
			if !ok {
				pullInfos = make([]dto.PullInfo, 0, 5)
			}
			pullInfo := dto.PullInfo{}
			pullInfo.UpId = v.TrafficOnId
			pullInfo.SourceId = v.SourceId
			pullInfo.OffId = v.TrafficOffId
			pullInfo.MediaId = v.MediaId
			pullInfo.PullUrl = v.CustomOutUrl
			pullInfo.OriginPullUrl = v.OutUrl
			pullInfos = append(pullInfos, pullInfo)
			pullInfoMap[strconv.Itoa(v.TrafficOffId)] = pullInfos
		}
	}

	//获取审核任务
	auditTasks := make([]live_schedule.AuditTask, 0, 5)
	err = e.Orm.Model(live_schedule.AuditTask{}).Where("tenant_id=? and channel_id=?", c.TenantId, c.ChannelId).Find(&auditTasks).Error
	if err != nil {
		e.Log.Errorf("Service GetWorkflow getChannelTask error:%s \r\n", err)
		return res, err
	}
	for _, v := range auditTasks {
		off, ok := offMap[v.TrafficOffId]
		if off.StreamType == 1 {
			pushInfos, ok := pushInfoMap[strconv.Itoa(v.TrafficOffId)]
			if !ok {
				pushInfos = make([]dto.PushInfo, 0, 5)
			}
			pushInfo := dto.PushInfo{}
			pushInfo.UpId = v.TrafficOnId
			pushInfo.SourceId = v.SourceId
			pushInfo.OffId = v.TrafficOffId
			pushInfo.MediaId = v.MediaId
			pushInfo.PushUrl = v.PushUrl
			pushInfos = append(pushInfos, pushInfo)
			pushInfoMap[strconv.Itoa(v.TrafficOffId)] = pushInfos
		}
		if ok && off.StreamType == 2 && off.TrafficOffType == 2 {
			pullInfos, ok := pullInfoMap[strconv.Itoa(v.TrafficOffId)]
			if !ok {
				pullInfos = make([]dto.PullInfo, 0, 5)
			}
			pullInfo := dto.PullInfo{}
			pullInfo.UpId = v.TrafficOnId
			pullInfo.SourceId = v.SourceId
			pullInfo.OffId = v.TrafficOffId
			pullInfo.MediaId = v.MediaId
			pullInfo.PullUrl = v.CustomOutUrl
			pullInfo.OriginPullUrl = v.OutUrl
			pullInfos = append(pullInfos, pullInfo)
			pullInfoMap[strconv.Itoa(v.TrafficOffId)] = pullInfos
		}
	}

	workflowDowns := make([]dto.WorkflowDown, 0, 5)
	for _, v := range offs {
		workflowDown := dto.WorkflowDown{}
		workflowDown.TrafficOff = v
		state := 0
		ok := false
		if len(tasks) > 0 {
			state, ok = offChannelStateMap[v.Id]
		} else {
			state, ok = offCortexStateMap[v.Id]
		}

		if ok {
			workflowDown.Status = strconv.Itoa(state)
		} else {
			if isAllRed {
				workflowDown.Status = "2"
			} else {
				workflowDown.Status = "1"
			}
		}
		if v.TrafficOffState == 2 {
			workflowDown.Status = "0"
		}
		workflowDown.PullInfos = pullInfoMap[strconv.Itoa(v.Id)]
		workflowDown.PushInfos = pushInfoMap[strconv.Itoa(v.Id)]
		workflowDowns = append(workflowDowns, workflowDown)
	}
	res.WorkflowDown = &workflowDowns

	return res, nil

}

func (e *ChannelTaskService) StartWorkflow(c *dto.ProcessWorkflowReq) error {

	var wg sync.WaitGroup
	now := time.Now().Unix()
	//if c.IsKeepEnable == 2 && c.EndTime <= now {
	//	return errors.New("结束时间不能小于当前时间")
	//}
	//设置矩阵为启用状态的任务id
	//ids := make([]int, 0, 5)
	streamHubService := &streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	baiduService := &baiduService.BaiduService{}
	baiduService.Log = e.Log
	cortexTaskService := &CortexTaskService{}
	cortexTaskService.Log = e.Log
	cortexTaskService.Orm = e.Orm
	auditTaskService := &AuditTaskService{}
	auditTaskService.Log = e.Log
	auditTaskService.Orm = e.Orm
	recordTaskService := &RecordTaskService{}
	recordTaskService.Log = e.Log
	recordTaskService.Orm = e.Orm
	convertTaskService := &ConvertTaskService{}
	convertTaskService.Log = e.Log
	convertTaskService.Orm = e.Orm
	transCodeTaskService := &TransCodeTaskService{}
	transCodeTaskService.Log = e.Log
	transCodeTaskService.Orm = e.Orm
	sourceBasicService := &SourceBasicService{}
	sourceBasicService.Log = e.Log
	sourceBasicService.Orm = e.Orm
	waterMarkTaskService := &WaterMarkTaskService{}
	waterMarkTaskService.Log = e.Log
	waterMarkTaskService.Orm = e.Orm
	mediaProcService := &mediaProcService.MediaProcService{}
	mediaProcService.Log = e.Log
	mediaProcService.Orm = e.Orm
	liveService := liveService.LiveBasic{}
	liveService.Log = e.Log
	liveService.Orm = e.Orm
	channelStatisticsService := &ChannelStatisticsService{}
	channelStatisticsService.Log = e.Log
	channelStatisticsService.Orm = e.Orm
	//获取通道基本信息
	var basic live_schedule.ChannelBasic
	e.Orm.Model(live_schedule.ChannelBasic{}).Where("id=?", c.ChannelId).Find(&basic)

	if c.StartType != 1 {
		err := e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", c.ChannelId).Update("terminal_time", 0).Error
		if err != nil {
			return err
		}
	}

	if basic.ChannelState == 1 || basic.ChannelState == 3 {
		return errors.New("通道已开启")
	}

	//关闭通道后需要30s之后才能开启通道
	if basic.LastStopTime+30 > now {
		return errors.New("正在回收资源，请 30 秒后重试")
	}

	//if basic.ChannelState == 4 {
	//	return errors.New("通道任务正在处理中")
	//}

	////设置为执行中
	//param1 := make(map[string]interface{})
	//param1["channel_state"] = 4
	//param1["progress"] = "0"
	//param1["update_time"] = now
	//err := e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", c.ChannelId).Updates(&param1).Error
	//if err != nil {
	//	e.Log.Errorf("update ChannelBasic error:%s \r\n", err)
	//	return err
	//}

	//获取媒体处理
	mediaProcess := make([]live_schedule.MediaProcess, 0, 2)
	err := e.Orm.Model(live_schedule.MediaProcess{}).Where("channel_id=?", c.ChannelId).Find(&mediaProcess).Error
	if err != nil {
		e.Log.Errorf("get MediaProcess error:%s \r\n", err)
		return err
	}

	//获取媒体处理
	upMedias := make([]live_schedule.UpMedia, 0, 5)
	err = e.Orm.Model(live_schedule.UpMedia{}).Where("channel_id=?", c.ChannelId).Find(&upMedias).Error
	if err != nil {
		e.Log.Errorf("get upMedia error:%s \r\n", err)
		//tx.Rollback()
		return err
	}
	upIds := make([]int, 0, 5)
	for _, v := range upMedias {
		upIds = append(upIds, v.UpId)
	}

	//获取媒体处理
	mediaOffs := make([]live_schedule.MediaOff, 0, 5)
	err = e.Orm.Model(live_schedule.MediaOff{}).Where("channel_id=?", c.ChannelId).Find(&mediaOffs).Error
	if err != nil {
		e.Log.Errorf("get MediaOff error:%s \r\n", err)
		//tx.Rollback()
		return err
	}
	offIds := make([]int, 0, 5)
	for _, v := range mediaOffs {
		offIds = append(offIds, v.OffId)
	}

	chartData := live_schedule.ChartData{}
	if len(basic.ChartData) > 0 {
		err = json.Unmarshal([]byte(basic.ChartData), &chartData)
		if err != nil {
			e.Log.Errorf("Unmarshal ChartData error:%s \r\n", err)
			return err
		}
	}

	sourceIds := make([]int, 0, 5)
	for k, datum := range chartData.NodeData {
		*(&chartData.NodeData[k].Node.Data.NodeState) = 1
		if datum.Node.Data.Type == "input" && datum.Node.Data.SourceFrom == "2" {
			sourceIds = append(sourceIds, datum.Node.Data.SourceId)
		}
	}
	if len(chartData.NodeData) > 0 {
		chartDataByte, err := json.Marshal(&chartData)
		if err != nil {
			e.Log.Errorf("Marshal chartData error:%s \r\n", err)
			return err
		}
		err = e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", c.ChannelId).Update("chart_data", string(chartDataByte)).Error
		if err != nil {
			e.Log.Errorf("update ChannelBasic error:%s \r\n", err)
			return err
		}
	}

	//获取上车点的信源

	//err = e.Orm.Model(live_schedule.TrafficOn{}).Distinct("source_id").Select("source_id").Where("channel_id = ?", c.ChannelId).Find(&sourceIds).Error
	//if err != nil {
	//	e.Log.Errorf("get TrafficOn error:%s \r\n", err)
	//	//tx.Rollback()
	//	return err
	//}

	//获取信源
	sources := make([]live_schedule.SourceBasic, 0, 5)
	err = e.Orm.Model(live_schedule.SourceBasic{}).Where("tenant_id=? and id in ?", c.TenantId, sourceIds).Find(&sources).Error
	if err != nil {
		e.Log.Errorf("ProcessWorkflow get SourceBasic error:%s \r\n", err)
		//tx.Rollback()
		return err
	}
	sourceMap := make(map[int]live_schedule.SourceBasic)
	for _, v := range sources {
		v := v
		sourceMap[v.Id] = v
	}

	//更新通道状态
	param := make(map[string]interface{})
	param["channel_state"] = 1
	//param["is_time"] = 2
	param["enable_time"] = now
	param["update_time"] = now
	if c.StartType == 1 {
		param["terminal_time"] = now
	}

	//param["start_time"] = now
	//if c.IsKeepEnable == 1 {
	//	param["repeat_end_time"] = 0
	//} else {
	//	param["repeat_end_time"] = c.EndTime
	//}
	err = e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", c.ChannelId).Updates(param).Error
	if err != nil {
		e.Log.Errorf("ChannelBasic Updates error:%s \r\n", err)
	}

	switchIds := make([]string, 0, 5)
	err = e.Orm.Model(&live_schedule.CortexOutput{}).Select("switch_task_id").Where("channel_id=? and switch_task_id!=''", c.ChannelId).Find(&switchIds).Error
	if err != nil {
		if err != nil {
			e.Log.Errorf("get CortexOutput error:%s \r\n", err)
			return err
		}
	}

	for _, v := range switchIds {
		if len(v) > 0 {
			err = streamHubService.StartSwitchSource(&StreamHubDto.StartOrStopSwitchSourceReq{Id: v, TenantId: basic.TenantId})
			if err != nil {
				e.Log.Errorf("StartSwitchSource error:%s \r\n", err)
				return err
			}
		}

	}
	channelTasks := make([]live_schedule.ChannelTask, 0, 5)
	cortexTasks := make([]live_schedule.CortexTask, 0, 5)
	auditTasks := make([]live_schedule.AuditTask, 0, 5)
	sourceRecordTasks := make([]live_schedule.SourceRecordTask, 0, 5)
	transCodeTasks := make([]live_schedule.ChannelTransCodeTask, 0, 5)
	waterMarkTransTasks := make([]live_schedule.WaterMarkTransTask, 0, 5)
	workNum := 0
	var partNum int
	sourceAuditMap := make(map[int]struct{})
	mediaMap := make(map[int]live_schedule.MediaProcess)
	mediaIds := make([]int, 0, 3)
	for _, m := range mediaProcess {
		m := m
		mediaIds = append(mediaIds, m.Id)
		mediaMap[m.MediaType] = m
		if m.MediaType == 1 {
			//获取频道下面所有的配置
			err = e.Orm.Model(live_schedule.ChannelTask{}).Where("tenant_id=? and channel_id=?", c.TenantId, c.ChannelId).Find(&channelTasks).Error
			if err != nil {
				e.Log.Errorf("ProcessWorkflow get ChannelTask error:%s \r\n", err)
				return err
			}
			workNum = workNum + len(channelTasks)
		} else if m.MediaType == 2 {
			//获取频道下面所有的配置
			err = e.Orm.Model(live_schedule.CortexTask{}).Where("tenant_id=? and channel_id=? and task_id!=''", c.TenantId, c.ChannelId).Find(&cortexTasks).Error
			if err != nil {
				e.Log.Errorf(" get CortexTask error:%s \r\n", err)
				return err
			}
			workNum = workNum + len(cortexTasks)
		} else if m.MediaType == 3 {
			//获取频道下面所有的审核配置
			err = e.Orm.Model(live_schedule.AuditTask{}).Where("tenant_id=? and channel_id=?", c.TenantId, c.ChannelId).Find(&auditTasks).Error
			if err != nil {
				e.Log.Errorf(" get CortexTask error:%s \r\n", err)
				return err
			}
			for _, v := range auditTasks {
				if len(v.AuditTaskId) > 0 {
					sourceAuditMap[v.SourceId] = struct{}{}
				}
			}
			workNum = workNum + len(auditTasks)
		} else if m.MediaType == 4 {
			//获取频道下面所有的审核配置
			err = e.Orm.Model(live_schedule.ChannelTransCodeTask{}).Where("tenant_id=? and channel_id=?", c.TenantId, c.ChannelId).Find(&transCodeTasks).Error
			if err != nil {
				e.Log.Errorf(" get TransCodeTask error:%s \r\n", err)
				return err
			}
			workNum = workNum + len(transCodeTasks)
		} else if m.MediaType == 5 {
			//获取频道下面所有的录制配置
			err = e.Orm.Model(live_schedule.SourceRecordTask{}).Where("tenant_id=? and channel_id=?", c.TenantId, c.ChannelId).Find(&sourceRecordTasks).Error
			if err != nil {
				e.Log.Errorf(" get SourceRecordTask error:%s \r\n", err)
				return err
			}
			workNum = workNum + len(sourceRecordTasks)
		} else if m.MediaType == 6 {
			//获取频道下面所有的水印配置
			err = e.Orm.Model(live_schedule.WaterMarkTransTask{}).Where("tenant_id=? and channel_id=?", c.TenantId, c.ChannelId).Find(&waterMarkTransTasks).Error
			if err != nil {
				e.Log.Errorf(" get WaterMarkTransTask error:%s \r\n", err)
				return err
			}
			workNum = workNum + len(waterMarkTransTasks)
		}
	}

	basics := make([]live_schedule.SourceBasic, 0, 5)
	for _, v := range sourceMap {
		v := v
		if (v.Kind == 2 || v.Kind == 3) && v.SourceState == 2 {
			basics = append(basics, v)
			//err = sourceBasicService.StartSource(&dto.StartSourceReq{TenantId: v.TenantId, Id: v.Id})
			//if err != nil {
			//	e.Log.Errorf("StartSource error:%s \r\n", err)
			//}
		}
	}

	if len(basics) > 0 {
		workNum = workNum + len(basics)
		e.StartOrStopCustomSources(1, basics, workNum, &partNum, param, c.ChannelId, sourceBasicService)
	}

	//if workNum == 0 {
	//	err = e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", basic.Id).Update("channel_state", 1).Error
	//	if err != nil {
	//		return err
	//	}
	//	return nil
	//}

	switchTasks := make([]live_schedule.SwitchTask, 0, 5)
	err = e.Orm.Model(&live_schedule.SwitchTask{}).Where("channel_id=?", c.ChannelId).Find(&switchTasks).Error
	if err != nil {
		e.Log.Errorf("StartSwitchSource error:%s \r\n", err)
		return err
	}
	for _, switchTask := range switchTasks {
		//开启协助任务
		err = streamHubService.StartSwitchSource(&StreamHubDto.StartOrStopSwitchSourceReq{TenantId: switchTask.TenantId, Id: switchTask.TaskId})
		if err != nil {
			e.Log.Errorf("StartStreamHub error:%s \r\n", err)
			return err
		}
	}

	channelTransTasks := make([]live_schedule.ChannelTransTask, 0, 5)
	err = e.Orm.Model(&live_schedule.ChannelTransTask{}).Where("channel_id=?", c.ChannelId).Find(&channelTransTasks).Error
	if err != nil {
		e.Log.Errorf("get ChannelTransTask error:%s \r\n", err)
		return err
	}
	for _, channelTransTask := range channelTransTasks {
		channelTransTask := channelTransTask
		go func() {
			e.Log.Trace("1111111111")
			//开启协助任务
			err = streamHubService.StartTransCode(&StreamHubDto.TransCodeReq{TenantId: channelTransTask.TenantId, Id: channelTransTask.TaskId})
			if err != nil {
				e.Log.Errorf("StartTransTask error:%s \r\n", err)
				//tx.Rollback()
				//return err
			}
			e.Log.Trace("jahsdjahdjahdjahdjad")
			//创建调度统计记录
			err = channelStatisticsService.CreateScheduleRecord(channelTransTask.TenantId, channelTransTask.ChannelId, channelTransTask.TaskId, 2, now)
			if err != nil {
				e.Log.Errorf("CreateScheduleRecord error:%s \r\n", err)
			}
		}()

	}

	covertTasks := make([]live_schedule.CovertTask, 0, 5)
	err = e.Orm.Model(&live_schedule.CovertTask{}).Where("channel_id=?", c.ChannelId).Find(&covertTasks).Error
	if err != nil {
		e.Log.Errorf("get CovertTask error:%s \r\n", err)
		return err
	}
	for _, covertTask := range covertTasks {
		//开启协助任务
		if len(covertTask.TaskId) > 0 {
			err = streamHubService.StartStreamHub(&StreamHubDto.StreamHubReq{TenantId: covertTask.TenantId, Id: covertTask.TaskId})
			if err != nil {
				e.Log.Errorf("StartStreamHub error:%s \r\n", err)
				//tx.Rollback()
				//return err
			}
			//创建调度统计记录
			err = channelStatisticsService.CreateScheduleRecord(covertTask.TenantId, covertTask.ChannelId, covertTask.TaskId, 1, now)
			if err != nil {
				e.Log.Errorf("CreateScheduleRecord error:%s \r\n", err)
			}
		}
	}

	if len(channelTasks) > 0 {
		convertTaskService.StartChannelTasks(&wg, channelTasks, workNum, &partNum, param, sourceMap)
	}

	if len(cortexTasks) > 0 {
		cortexTaskService.StartCortexTasks(&wg, cortexTasks, workNum, &partNum, param, sourceMap)
	}

	if len(auditTasks) > 0 {
		m := mediaMap[3]
		auditTaskService.StartAuditTasks(&wg, auditTasks, m, sourceAuditMap, workNum, &partNum, param, sourceMap)
	}

	if len(sourceRecordTasks) > 0 {
		recordTaskService.StartRecordTasks(&wg, sourceRecordTasks, workNum, &partNum, param, sourceMap)
	}
	if len(transCodeTasks) > 0 {
		transCodeTaskService.StartTransCodeTasks(&wg, transCodeTasks, workNum, &partNum, param, sourceMap)
	}
	if len(waterMarkTransTasks) > 0 {
		waterMarkTaskService.StartWatermarkTasks(&wg, waterMarkTransTasks, workNum, &partNum, param, sourceMap)
	}

	err = e.Orm.Model(&live_schedule.TrafficOn{}).Where("id in ?", upIds).Update("traffic_on_state", 1).Error
	if err != nil {
		e.Log.Errorf(" update TrafficOn error:%s \r\n", err)
		return err
	}
	err = e.Orm.Model(&live_schedule.MediaProcess{}).Where("id in ?", mediaIds).Update("media_state", 1).Error
	if err != nil {
		e.Log.Errorf(" update MediaProcess error:%s \r\n", err)
		return err
	}

	err = e.Orm.Model(&live_schedule.TrafficOff{}).Where("id in ?", offIds).Update("traffic_off_state", 1).Error
	if err != nil {
		e.Log.Errorf(" update TrafficOff error:%s \r\n", err)
		return err
	}

	//立即开始会把当天的定时任务做废掉
	//cycleTimes := make([]live_schedule.CycleTime, 0, 3)
	//if c.StartType == 1 && basic.IsTime == 1 {
	//	err = e.Orm.Model(&live_schedule.CycleTime{}).Where("channel_id=? and time_type=? and start_time>? and is_cancel=2", c.ChannelId, basic.TimeType, now).Find(&cycleTimes).Error
	//	if err != nil {
	//		e.Log.Errorf(" get CycleTime error:%s \r\n", err)
	//		return err
	//	}
	//
	//	//if len(cycleTimes) > 0 {
	//	//	startTime := time.Unix(cycleTimes[0].StartTime, 0)
	//	//	now1 := time.Now()
	//	//	if startTime.Year() == now1.Year() && startTime.Month() == now1.Month() && startTime.Day() == now1.Day() {
	//	//		err = e.Orm.Model(&live_schedule.CycleTime{}).Where("id=?", cycleTimes[0].Id).Update("is_cancel", 1).Error
	//	//		if err != nil {
	//	//			e.Log.Errorf(" update CycleTime error:%s \r\n", err)
	//	//			return err
	//	//		}
	//	//	}
	//	//
	//	//}
	//	//if basic.TimeType == 1 {
	//	//	param := make(map[string]interface{})
	//	//	param["is_time"] = 2
	//	//	param["start_time"] = 0
	//	//	param["end_time"] = 0
	//	//	param["time_type"] = 1
	//	//	param["cycle_end_time"] = 0
	//	//	param["weekly"] = ""
	//	//	err = e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", basic.Id).Updates(&param).Error
	//	//	if err != nil {
	//	//		e.Log.Errorf("update ChannelBasic error:%s \r\n", err)
	//	//	}
	//	//}
	//}

	wg.Wait()
	return nil
}

func (e *ChannelTaskService) StopWorkflow(c *dto.ProcessWorkflowReq) error {

	var wg sync.WaitGroup
	liveService := &liveService.LiveBasic{}
	streamHubService := &streamHubService.StreamHubService{}
	liveService.Log = e.Log
	streamHubService.Log = e.Log
	baiduService := &baiduService.BaiduService{}
	baiduService.Log = e.Log
	sourceBasicService := &SourceBasicService{}
	sourceBasicService.Log = e.Log
	sourceBasicService.Orm = e.Orm
	cortexTaskService := &CortexTaskService{}
	cortexTaskService.Log = e.Log
	cortexTaskService.Orm = e.Orm
	auditTaskService := &AuditTaskService{}
	auditTaskService.Log = e.Log
	auditTaskService.Orm = e.Orm
	recordTaskService := &RecordTaskService{}
	recordTaskService.Log = e.Log
	recordTaskService.Orm = e.Orm
	convertTaskService := &ConvertTaskService{}
	convertTaskService.Log = e.Log
	convertTaskService.Orm = e.Orm
	transCodeTaskService := &TransCodeTaskService{}
	transCodeTaskService.Log = e.Log
	transCodeTaskService.Orm = e.Orm
	waterMarkTaskService := &WaterMarkTaskService{}
	waterMarkTaskService.Log = e.Log
	waterMarkTaskService.Orm = e.Orm
	mediaProcService := &mediaProcService.MediaProcService{}
	mediaProcService.Log = e.Log
	mediaProcService.Orm = e.Orm

	channelTasks := make([]live_schedule.ChannelTask, 0, 5)
	cortexTasks := make([]live_schedule.CortexTask, 0, 5)
	auditTasks := make([]live_schedule.AuditTask, 0, 5)
	sourceRecordTasks := make([]live_schedule.SourceRecordTask, 0, 5)
	transCodeTasks := make([]live_schedule.ChannelTransCodeTask, 0, 5)
	waterMarkTransTasks := make([]live_schedule.WaterMarkTransTask, 0, 5)
	workNum := 0
	var partNum int
	now := time.Now().Unix()

	//获取媒体处理
	mediaProcess := make([]live_schedule.MediaProcess, 0, 5)
	err := e.Orm.Model(live_schedule.MediaProcess{}).Where("channel_id=?", c.ChannelId).Find(&mediaProcess).Error
	if err != nil {
		e.Log.Errorf("get MediaProcess error:%s \r\n", err)
		//tx.Rollback()
		return err
	}
	//获取通道基本信息
	var basic live_schedule.ChannelBasic
	e.Orm.Model(live_schedule.ChannelBasic{}).Where("id=?", c.ChannelId).Find(&basic)
	if basic.ChannelState == 2 {
		return errors.New("通道已停用")
	}

	//if basic.ChannelState == 4 {
	//	return errors.New("通道任务正在处理中")
	//}

	////设置为执行中
	//param1 := make(map[string]interface{})
	//param1["channel_state"] = 4
	//param1["progress"] = "0"
	//param1["update_time"] = now
	//err = e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", c.ChannelId).Updates(&param1).Error
	//if err != nil {
	//	e.Log.Errorf("update ChannelBasic error:%s \r\n", err)
	//	return err
	//}

	chartData := live_schedule.ChartData{}
	if len(basic.ChartData) > 0 {
		err = json.Unmarshal([]byte(basic.ChartData), &chartData)
		if err != nil {
			e.Log.Errorf("Unmarshal ChartData error:%s \r\n", err)
			return err
		}
	}

	sourceIds := make([]int, 0, 5)
	for k, datum := range chartData.NodeData {
		*(&chartData.NodeData[k].Node.Data.NodeState) = 2
		if datum.Node.Data.Type == "input" && datum.Node.Data.SourceFrom == "2" {
			sourceIds = append(sourceIds, datum.Node.Data.SourceId)
		}
	}

	switchIds := make([]string, 0, 5)
	err = e.Orm.Model(&live_schedule.CortexOutput{}).Select("switch_task_id").Where("channel_id=?", c.ChannelId).Find(&switchIds).Error
	if err != nil {
		if err != nil {
			e.Log.Errorf("get CortexOutput error:%s \r\n", err)
			return err
		}
	}
	//停用切换任务
	for _, v := range switchIds {
		if len(v) > 0 {
			err = streamHubService.StopSwitchSource(&StreamHubDto.StartOrStopSwitchSourceReq{Id: v, TenantId: basic.TenantId})
			if err != nil {
				e.Log.Errorf("StopSwitchSource error:%s \r\n", err)
				return err
			}
		}

	}

	////获取上车点的信源
	//sourceIds := make([]int, 0, 5)
	//err = e.Orm.Model(live_schedule.TrafficOn{}).Distinct("source_id").Select("source_id").Where("channel_id=?", c.ChannelId).Find(&sourceIds).Error
	//if err != nil {
	//	e.Log.Errorf("get TrafficOn error:%s \r\n", err)
	//	return err
	//}

	//获取信源
	sources := make([]live_schedule.SourceBasic, 0, 5)
	err = e.Orm.Model(live_schedule.SourceBasic{}).Where("tenant_id=? and id in ?", c.TenantId, sourceIds).Find(&sources).Error
	if err != nil {
		e.Log.Errorf("ProcessWorkflow get SourceBasic error:%s \r\n", err)
		return err
	}
	sourceMap := make(map[int]live_schedule.SourceBasic)
	for _, v := range sources {
		v := v
		sourceMap[v.Id] = v
	}
	//更新通道状态
	param := make(map[string]interface{})
	param["update_time"] = now
	//param["is_time"] = 2
	param["channel_state"] = 2
	param["last_stop_time"] = now
	//param["end_time"] = now

	basics := make([]live_schedule.SourceBasic, 0, 3)
	for _, v := range sourceMap {
		v := v
		if (v.Kind == 2 || v.Kind == 3) && v.SourceState != 2 {
			basics = append(basics, v)

			//err = sourceBasicService.StopSource(&dto.StopSourceReq{TenantId: v.TenantId, Id: v.Id, TaskId: v.TaskId})
			//if err != nil {
			//	e.Log.Errorf("StopSource error:%s \r\n", err)
			//}
		}
	}
	if len(basics) > 0 {
		workNum = workNum + len(basics)
		e.StartOrStopCustomSources(2, basics, workNum, &partNum, param, c.ChannelId, sourceBasicService)
	}

	//汇聚协转
	for _, m := range mediaProcess {
		if m.MediaType == 1 {
			//获取频道下面所有的配置
			err = e.Orm.Model(live_schedule.ChannelTask{}).Where("tenant_id=? and channel_id=?", c.TenantId, c.ChannelId).Find(&channelTasks).Error
			if err != nil {
				e.Log.Errorf("ProcessWorkflow get ChannelTask error:%s \r\n", err)
				return err
			}
			workNum = workNum + len(channelTasks)

		} else if m.MediaType == 2 {
			//获取频道下面所有的配置
			err = e.Orm.Model(live_schedule.CortexTask{}).Where("tenant_id=? and channel_id=? and task_id!=''", c.TenantId, c.ChannelId).Find(&cortexTasks).Error
			if err != nil {
				e.Log.Errorf(" get CortexTask error:%s \r\n", err)
				return err
			}
			workNum = workNum + len(cortexTasks)

		} else if m.MediaType == 3 {
			//获取频道下面所有的审核配置
			err = e.Orm.Model(live_schedule.AuditTask{}).Where("tenant_id=? and channel_id=?", c.TenantId, c.ChannelId).Find(&auditTasks).Error
			if err != nil {
				e.Log.Errorf(" get AuditTask error:%s \r\n", err)
				return err
			}
			workNum = workNum + len(auditTasks)

		} else if m.MediaType == 4 {
			//获取频道下面所有的转码配置
			err = e.Orm.Model(live_schedule.ChannelTransCodeTask{}).Where("tenant_id=? and channel_id=?", c.TenantId, c.ChannelId).Find(&transCodeTasks).Error
			if err != nil {
				e.Log.Errorf(" get ChannelTransCodeTask error:%s \r\n", err)
				return err
			}
			workNum = workNum + len(transCodeTasks)

		} else if m.MediaType == 5 {
			//获取频道下面所有的录制配置
			err = e.Orm.Model(live_schedule.SourceRecordTask{}).Where("tenant_id=? and channel_id=?", c.TenantId, c.ChannelId).Find(&sourceRecordTasks).Error
			if err != nil {
				e.Log.Errorf(" get SourceRecordTask error:%s \r\n", err)
				return err
			}
			workNum = workNum + len(sourceRecordTasks)
		} else if m.MediaType == 6 {
			//获取频道下面所有的水印配置
			err = e.Orm.Model(live_schedule.WaterMarkTransTask{}).Where("tenant_id=? and channel_id=?", c.TenantId, c.ChannelId).Find(&waterMarkTransTasks).Error
			if err != nil {
				e.Log.Errorf(" get WaterMarkTransTask error:%s \r\n", err)
				return err
			}
			workNum = workNum + len(waterMarkTransTasks)
		}
	}
	//if workNum == 0 {
	//	err = e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", basic.Id).Update("channel_state", 2).Error
	//	if err != nil {
	//		return err
	//	}
	//	return nil
	//}

	switchTasks := make([]live_schedule.SwitchTask, 0, 5)
	err = e.Orm.Model(&live_schedule.SwitchTask{}).Where("channel_id=?", c.ChannelId).Find(&switchTasks).Error
	if err != nil {
		e.Log.Errorf("StartSwitchSource error:%s \r\n", err)
		return err
	}
	for _, switchTask := range switchTasks {
		//开启协助任务
		err := streamHubService.StopSwitchSource(&StreamHubDto.StartOrStopSwitchSourceReq{TenantId: switchTask.TenantId, Id: switchTask.TaskId})
		if err != nil {
			e.Log.Errorf("StopSwitchSource error:%s \r\n", err)
			//tx.Rollback()
			//return err
		}
	}

	covertTasks := make([]live_schedule.CovertTask, 0, 5)
	err = e.Orm.Model(&live_schedule.CovertTask{}).Where("channel_id=?", c.ChannelId).Find(&covertTasks).Error
	if err != nil {
		e.Log.Errorf("get CovertTask error:%s \r\n", err)
		return err
	}
	for _, covertTask := range covertTasks {
		//停止协助任务
		err = streamHubService.StopStreamHub(&StreamHubDto.StreamHubReq{TenantId: covertTask.TenantId, Id: covertTask.TaskId})
		if err != nil {
			e.Log.Errorf("StopStreamHub error:%s \r\n", err)
			//tx.Rollback()
			//return err
		}
		//更新任务记录结束时间
		param2 := make(map[string]interface{})
		param2["end_time"] = now
		err = e.Orm.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id = ? and end_time = 0 and type =1", covertTask.TaskId).Updates(&param2).Error
		if err != nil {
			e.Log.Errorf("Updates ScheduleRecord error:%s \r\n", err)
		}
	}

	channelTransTasks := make([]live_schedule.ChannelTransTask, 0, 5)
	err = e.Orm.Model(&live_schedule.ChannelTransTask{}).Where("channel_id=?", c.ChannelId).Find(&channelTransTasks).Error
	if err != nil {
		e.Log.Errorf("get ChannelTransTask error:%s \r\n", err)
		return err
	}
	for _, channelTransTask := range channelTransTasks {
		//开启协助任务
		err = streamHubService.StopTransCode(&StreamHubDto.TransCodeReq{TenantId: channelTransTask.TenantId, Id: channelTransTask.TaskId})
		if err != nil {
			e.Log.Errorf("StopTransCode error:%s \r\n", err)
			//tx.Rollback()
			//return err
		}
		//更新任务记录结束时间
		param2 := make(map[string]interface{})
		param2["end_time"] = now
		err = e.Orm.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id = ? and end_time = 0 and type =2", channelTransTask.TaskId).Updates(&param2).Error
		if err != nil {
			e.Log.Errorf("Updates ScheduleRecord error:%s \r\n", err)
		}
	}

	if len(channelTasks) > 0 {
		convertTaskService.StopChannelTasks(&wg, channelTasks, workNum, &partNum, param, sourceMap)
	}

	if len(cortexTasks) > 0 {
		cortexTaskService.StopCortexTasks(&wg, cortexTasks, workNum, &partNum, param, sourceMap)
	}

	if len(auditTasks) > 0 {
		auditTaskService.StopAuditTasks(&wg, auditTasks, workNum, &partNum, param, sourceMap)
	}

	if len(sourceRecordTasks) > 0 {
		recordTaskService.StopRecordTasks(&wg, sourceRecordTasks, workNum, &partNum, param, sourceMap)
	}
	if len(transCodeTasks) > 0 {
		transCodeTaskService.StopTransCodeTasks(&wg, transCodeTasks, workNum, &partNum, param, sourceMap)
	}
	if len(waterMarkTransTasks) > 0 {
		waterMarkTaskService.StopWaterMarkTasks(&wg, waterMarkTransTasks, workNum, &partNum, param, sourceMap)
	}

	err = e.Orm.Model(&live_schedule.TrafficOn{}).Where("channel_id = ?", c.ChannelId).Update("traffic_on_state", 2).Error
	if err != nil {
		e.Log.Errorf(" update TrafficOn error:%s \r\n", err)
		return err
	}

	err = e.Orm.Model(&live_schedule.MediaProcess{}).Where("channel_id = ?", c.ChannelId).Update("media_state", 2).Error
	if err != nil {
		e.Log.Errorf(" update MediaProcess error:%s \r\n", err)
		return err
	}

	err = e.Orm.Model(&live_schedule.TrafficOff{}).Where("channel_id = ?", c.ChannelId).Update("traffic_off_state", 2).Error
	if err != nil {
		e.Log.Errorf(" update TrafficOff error:%s \r\n", err)
		return err
	}

	if c.TimeTaskId > 0 {
		err = e.Orm.Model(&live_schedule.CycleTime{}).Where("id =?", c.TimeTaskId).Update("is_cancel", 1).Error
		if err != nil {
			e.Log.Errorf("update CycleTime error:%s \r\n", err)
			return err
		}
	}

	err = e.Orm.Model(live_schedule.ChannelBasic{}).Where("id=?", c.ChannelId).Updates(param).Error
	if err != nil {
		e.Log.Errorf("Update ChannelTask error:%s \r\n", err)
	}

	if c.StartType == 1 && basic.IsTime == 1 {
		err = e.Orm.Model(&live_schedule.CycleTime{}).Where("channel_id =? and time_type=? and is_cancel=2 and start_time<? and end_time>?", c.ChannelId, basic.TimeType, now, now).Update("is_cancel", 1).Error
		if err != nil {
			e.Log.Errorf("update CycleTime error:%s \r\n", err)
			return err
		}

		if basic.TimeType == 1 {
			//如果不重复任务 在任务期间停止 删除定制任务
			if basic.StartTime < now && basic.EndTime > now {
				param := make(map[string]interface{})
				param["is_time"] = 2
				param["start_time"] = 0
				param["end_time"] = 0
				param["time_type"] = 1
				param["cycle_end_time"] = 0
				param["weekly"] = ""
				err = e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", basic.Id).Updates(&param).Error
				if err != nil {
					e.Log.Errorf("update ChannelBasic error:%s \r\n", err)
				}
			}
		}
	}
	if len(chartData.NodeData) > 0 {
		chartDataByte, err := json.Marshal(&chartData)
		if err != nil {
			e.Log.Errorf("Marshal chartData error:%s \r\n", err)
			return err
		}
		err = e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", c.ChannelId).Update("chart_data", string(chartDataByte)).Error
		if err != nil {
			e.Log.Errorf("update ChannelBasic error:%s \r\n", err)
			return err
		}
	}

	wg.Wait()
	return nil
}

func (e *ChannelTaskService) TimeStartWorkflow() {
	// 使用 os.Stat 函数获取文件的信息
	_, err := os.Stat("/var/vs/failover/master")

	// 判断文件是否存在
	if os.IsNotExist(err) {
		e.Log.Errorf("文件不存在")
		return
	}
	//获取需要定时启动的通道
	basics := make([]live_schedule.ChannelBasic, 0, 5)
	err = e.Orm.Model(live_schedule.ChannelBasic{}).Where("is_time =1 and is_delete=2").Find(&basics).Error
	if err != nil {
		e.Log.Errorf("get ChannelBasic error:%s \r\n", err)
	}
	now := time.Now()
	for _, v := range basics {
		v := v
		switch v.TimeType {
		case 1:
			e.NoRepeatChannel(v, now)
		case 2:
			e.DailyChannel(v, now)
		case 3:
			e.WeeklyChannel(v, now)
		case 4:
			e.WeeklyChannel(v, now)
		default:
			e.NoRepeatChannel(v, now)
		}

		continue
	}
}

func (e *ChannelTaskService) NoRepeatChannel(v live_schedule.ChannelBasic, t time.Time) {

	now := t.Unix()
	//fmt.Println("KKKKKKKKKK1:", time.Unix(v.StartTime, 0), time.Unix(now, 0), v.StartTime > now)
	// 当前时间早于开启时间就跳过
	if v.StartTime > now {
		return
	}

	//当前时间介于开启和结束时间并且通道为关闭状态
	//fmt.Println("KKKKKKKKKK:", time.Unix(v.StartTime, 0), time.Unix(v.EndTime, 0), time.Unix(now, 0))
	if v.StartTime <= now && v.EndTime > now {
		//fmt.Println("BBBBBB:", time.Unix(v.StartTime, 0), time.Unix(v.EndTime, 0), time.Unix(now, 0))
		err := e.StartWorkflow(&dto.ProcessWorkflowReq{TenantId: v.TenantId, ChannelId: v.Id})
		if err != nil {
			e.Log.Errorf("StartWorkflow error:%s \r\n", err)
		}
	}
	//当前时间大于结束时间并且通道为开启状态
	if v.EndTime <= now && v.ChannelState != 2 {
		if v.TerminalTime > 0 {
			return
		}
		err := e.StopWorkflow(&dto.ProcessWorkflowReq{TenantId: v.TenantId, ChannelId: v.Id})
		if err != nil {
			e.Log.Errorf("StopWorkflow error:%s \r\n", err)
		}
		param := make(map[string]interface{})
		param["is_time"] = 2
		param["start_time"] = 0
		param["end_time"] = 0
		param["time_type"] = 1
		param["cycle_end_time"] = 0
		param["weekly"] = ""
		err = e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", v.Id).Updates(&param).Error
		if err != nil {
			e.Log.Errorf("update ChannelBasic error:%s \r\n", err)
		}
	}
}

func (e *ChannelTaskService) DailyChannel(v live_schedule.ChannelBasic, now time.Time) {

	//// 当前时间早于开启时间就跳过
	//if v.StartTime > now.Unix() {
	//	return
	//}

	endDate := time.Unix(v.EndTime, 0)
	//todayStartTime := time.Date(now.Year(), now.Month(), now.Day(), startDate.Hour(), startDate.Minute(), startDate.Second(), 0, now.Location()).Unix()
	//todayEndTime := time.Date(now.Year(), now.Month(), now.Day(), endDate.Hour(), endDate.Minute(), endDate.Second(), 0, now.Location()).Unix()
	cycleEndTime := time.Unix(v.CycleEndTime, 0)

	if v.CycleEndTime > 0 {
		v.CycleEndTime = time.Date(cycleEndTime.Year(), cycleEndTime.Month(), cycleEndTime.Day(), endDate.Hour(), endDate.Minute(), endDate.Second(), 0, now.Location()).Unix()
		if v.CycleEndTime < now.Unix() && v.ChannelState == 2 {

			param := make(map[string]interface{})
			param["is_time"] = 2
			param["start_time"] = 0
			param["end_time"] = 0
			param["time_type"] = 1
			param["cycle_end_time"] = 0
			param["weekly"] = ""
			err := e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", v.Id).Updates(&param).Error
			if err != nil {
				e.Log.Errorf("update ChannelBasic error:%s \r\n", err)
			}
			err = e.Orm.Model(&live_schedule.CycleTime{}).Where("channel_id=?", v.Id).Delete(&live_schedule.CycleTime{}).Error
			if err != nil {
				e.Log.Errorf("delete CycleTime error:%s \r\n", err)
			}
			return
		}
	}

	//year, month, day := now.Date()
	//startOfDay := time.Date(year, month, day, 0, 0, 0, 0, now.Location())
	//endOfDay := time.Date(year, month, day, 23, 59, 59, 0, now.Location())

	cycleTime := live_schedule.CycleTime{}
	//err := e.Orm.Model(&live_schedule.CycleTime{}).Where("start_time>? and start_time<? and channel_id=?", startOfDay.Unix(), endOfDay.Unix(), v.Id).Find(&cycleTime).Error
	//if err != nil {
	//	e.Log.Errorf("get CycleTime error:%s \r\n", err)
	//}

	err := e.Orm.Model(&live_schedule.CycleTime{}).Where("start_time<? and end_time>? and channel_id=? and time_type =?", now.Unix(), now.Unix(), v.Id, v.TimeType).Find(&cycleTime).Error
	if err != nil {
		e.Log.Errorf("get CycleTime error:%s \r\n", err)
	}

	//fmt.Println("KKKKKKKKKKKKKKKKKKK", time.Unix(cycleTime.StartTime, 0), "  qqqqqq:", cycleTime.IsCancel)
	//if (cycleTime.StartTime != todayStartTime || cycleTime.EndTime != todayEndTime) && cycleTime.Id > 0 {
	//	todayStartTime = cycleTime.StartTime
	//	todayEndTime = cycleTime.EndTime
	//}
	//
	////fmt.Println(cycleTime)
	if cycleTime.Id > 0 && cycleTime.IsCancel == 1 {
		return
	}
	//
	////当前时间介于开启和结束时间并且通道为关闭状态
	//if todayStartTime < now.Unix() && todayEndTime > now.Unix() && v.ChannelState == 2 {
	//	if todayStartTime < v.LastStopTime && todayEndTime > v.LastStopTime {
	//		return
	//	}
	//	err := e.StartWorkflow(&dto.ProcessWorkflowReq{TenantId: v.TenantId, ChannelId: v.Id})
	//	if err != nil {
	//		e.Log.Errorf("StartWorkflow error:%s \r\n", err)
	//	}
	//}
	////当前时间大于结束时间并且通道为开启状态
	//if todayEndTime < now.Unix() && v.ChannelState != 2 {
	//	err = e.StopWorkflow(&dto.ProcessWorkflowReq{TenantId: v.TenantId, ChannelId: v.Id, TimeTaskId: cycleTime.Id})
	//	if err != nil {
	//		e.Log.Errorf("StopWorkflow error:%s \r\n", err)
	//	}
	//}
	if cycleTime.Id > 0 {
		fmt.Println("定时任务开启:", cycleTime.Id, v.ChannelState)
		err = e.StartWorkflow(&dto.ProcessWorkflowReq{TenantId: v.TenantId, ChannelId: v.Id})
		if err != nil {
			e.Log.Errorf("StartWorkflow error:%s \r\n", err)
		}
	}
	if cycleTime.Id <= 0 && v.ChannelState == 1 {
		if v.TerminalTime > 0 {
			return
		}
		err = e.StopWorkflow(&dto.ProcessWorkflowReq{TenantId: v.TenantId, ChannelId: v.Id})
		if err != nil {
			e.Log.Errorf("StopWorkflow error:%s \r\n", err)
		}
	}
}

func (e *ChannelTaskService) WeeklyChannel(v live_schedule.ChannelBasic, now time.Time) {

	//// 当前时间早于开启时间就跳过
	//if v.StartTime > now.Unix() {
	//	return
	//}

	////当前星期不是设置的星期任务跳过
	//startWeek := time.Unix(v.StartTime, 0).Weekday()
	//todayWeek := time.Now().Weekday()
	//if startWeek != todayWeek {
	//	return
	//}

	startDate := time.Unix(v.StartTime, 0)
	endDate := time.Unix(v.EndTime, 0)

	todayStartTime := time.Date(now.Year(), now.Month(), now.Day(), startDate.Hour(), startDate.Minute(), startDate.Second(), 0, now.Location()).Unix()
	todayEndTime := time.Date(now.Year(), now.Month(), now.Day(), endDate.Hour(), endDate.Minute(), endDate.Second(), 0, now.Location()).Unix()

	if v.CycleEndTime != 0 {
		cycleEndTime := time.Unix(v.CycleEndTime, 0)
		var cycleTime live_schedule.CycleTime
		err := e.Orm.Model(&live_schedule.CycleTime{}).Where("channel_id=?", v.Id).Order("start_time desc").First(&cycleTime).Error
		if err != nil {
			e.Log.Errorf("get CycleTime error:%s \r\n", err)
		}
		cycleEndTime = time.Unix(cycleTime.EndTime, 0)
		v.CycleEndTime = time.Date(cycleEndTime.Year(), cycleEndTime.Month(), cycleEndTime.Day(), cycleEndTime.Hour(), cycleEndTime.Minute(), cycleEndTime.Second(), 0, now.Location()).Unix()
		//如果当前日期大于重复任务的结束时间退出
		if v.CycleEndTime < now.Unix() && v.ChannelState == 2 {
			param := make(map[string]interface{})
			param["is_time"] = 2
			param["start_time"] = 0
			param["end_time"] = 0
			param["time_type"] = 1
			param["cycle_end_time"] = 0
			param["weekly"] = ""
			err = e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", v.Id).Updates(&param).Error
			if err != nil {
				e.Log.Errorf("update ChannelBasic error:%s \r\n", err)
			}
			err = e.Orm.Model(&live_schedule.CycleTime{}).Where("channel_id=?", v.Id).Delete(&live_schedule.CycleTime{}).Error
			if err != nil {
				e.Log.Errorf("delete CycleTime error:%s \r\n", err)
			}
			return
		}
	}
	year, month, day := now.Date()
	startOfDay := time.Date(year, month, day, 0, 0, 0, 0, now.Location())
	endOfDay := time.Date(year, month, day, 23, 59, 59, 0, now.Location())

	fmt.Println(startOfDay, endOfDay)
	cycleTime := live_schedule.CycleTime{}
	err := e.Orm.Model(&live_schedule.CycleTime{}).Where("start_time<? and end_time>? and channel_id=? and time_type =?", now.Unix(), now.Unix(), v.Id, v.TimeType).Find(&cycleTime).Error
	if err != nil {
		e.Log.Errorf("get CycleTime error:%s \r\n", err)
	}
	if (cycleTime.StartTime != todayStartTime || cycleTime.EndTime != todayEndTime) && cycleTime.Id > 0 {
		todayStartTime = cycleTime.StartTime
		todayEndTime = cycleTime.EndTime
	}
	if cycleTime.Id > 0 && cycleTime.IsCancel == 1 {
		return
	}

	w := now.Weekday()
	repeatDays := make([]int, 0, 7)
	if len(v.Weekly) > 0 {
		arr := strings.Split(v.Weekly, ",")
		for _, week := range arr {
			week := week
			w, _ := strconv.Atoi(strings.Trim(week, " "))
			repeatDays = append(repeatDays, w)
		}
	}
	isPlan := false

	for _, r := range repeatDays {
		//判断开启是当天的是定时任务当天
		if r == int(w) || v.ChannelState == 1 {
			isPlan = true
			break
		}
	}

	//不在周内规定的星期则跳过
	if !isPlan {
		return
	}

	if cycleTime.Id > 0 {
		err = e.StartWorkflow(&dto.ProcessWorkflowReq{TenantId: v.TenantId, ChannelId: v.Id})
		if err != nil {
			e.Log.Errorf("StartWorkflow error:%s \r\n", err)
		}
	}
	if cycleTime.Id <= 0 {
		if v.TerminalTime > 0 {
			return
		}
		err = e.StopWorkflow(&dto.ProcessWorkflowReq{TenantId: v.TenantId, ChannelId: v.Id})
		if err != nil {
			e.Log.Errorf("StopWorkflow error:%s \r\n", err)
		}
	}

	////当前时间介于开启和结束时间并且通道为关闭状态
	//if todayStartTime < now.Unix() && todayEndTime > now.Unix() && v.ChannelState == 2 {
	//	if todayStartTime < v.LastStopTime && todayEndTime > v.LastStopTime {
	//		return
	//	}
	//	err = e.StartWorkflow(&dto.ProcessWorkflowReq{TenantId: v.TenantId, ChannelId: v.Id})
	//	if err != nil {
	//		e.Log.Errorf("StartWorkflow error:%s \r\n", err)
	//	}
	//}
	////当前时间大于结束时间并且通道为开启状态
	//if todayEndTime < now.Unix() && v.ChannelState != 2 {
	//	err = e.StopWorkflow(&dto.ProcessWorkflowReq{TenantId: v.TenantId, ChannelId: v.Id, TimeTaskId: cycleTime.Id})
	//	if err != nil {
	//		e.Log.Errorf("StopWorkflow error:%s \r\n", err)
	//	}
	//}
}

// ProcessBlackWhite 处理黑白名单
func (e *ChannelTaskService) ProcessBlackWhite(taskIds []string, off live_schedule.TrafficOff) error {

	//taskIds := make([]string, 0, 5)
	//err = tx.Model(&live_schedule.ChannelTask{}).Select("task_id").Where("off_id=?", data.Id).Find(&taskIds).Error
	//if err != nil {
	//	tx.Rollback()
	//	e.Log.Errorf("get ChannelTask error:%s \r\n", err)
	//	return err
	//}
	//
	//err = tx.Model(&live_schedule.HlsBlackWhite{}).Where("off_id=?", data.Id).Delete(&live_schedule.HlsBlackWhite{}).Error
	//if err != nil {
	//	tx.Rollback()
	//	e.Log.Errorf("delete HlsBlackWhite error:%s \r\n", err)
	//	return err
	//}
	streamHubService := &streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	params := make([]StreamHubDto.SetBlackWhiteParam, 0, 5)

	for _, v := range taskIds {
		v := v
		param := StreamHubDto.SetBlackWhiteParam{}
		param.TenantId = off.TenantId
		param.TaskId = v
		if off.TrafficOffType != 2 {
			continue
		}
		if off.EnableBlackWhite == 2 {
			param.Acl = append(param.Acl, "allow all")
		} else if off.ListType == 1 {
			for _, ip := range off.IPAddresses {
				param.Acl = append(param.Acl, "deny "+ip.IpAddress)
			}
		} else {
			for _, ip := range off.IPAddresses {
				param.Acl = append(param.Acl, "allow "+ip.IpAddress)
			}
		}
		params = append(params, param)
	}

	for _, v := range params {
		v := v
		if len(v.TaskId) > 0 {
			err := streamHubService.SetHlsWhiteBlack(&v)
			if err != nil {
				e.Log.Errorf("SetHlsWhiteBlack error:%s \r\n", err)
			}
		}
	}
	return nil
}

func (e *ChannelTaskService) StartOrStopCustomSources(kind int, tasks []live_schedule.SourceBasic, workNum int, partNum *int, param map[string]interface{}, channelId int, sourceBasicService *SourceBasicService) error {
	// 创建任务通道
	taskCh := make(chan live_schedule.SourceBasic, 3)
	// 启动协程执行任务
	go func() {
		for i := 0; i < len(tasks); i++ {
			taskCh <- tasks[i] // 将任务放入通道
		}
		close(taskCh) // 关闭通道，表示任务已经全部放入通道
	}()

	//var mu sync.Mutex
	for i := 0; i < 3; i++ {
		go func() {
			for v := range taskCh {
				//停止协助任务
				if len(v.TaskId) > 0 {
					if kind == 1 {
						err := sourceBasicService.StartSource(&dto.StartSourceReq{TenantId: v.TenantId, Id: v.Id})
						if err != nil {
							e.Log.Errorf("StartSource error:%s \r\n", err)
						}
					} else {
						err := sourceBasicService.StopSource(&dto.StopSourceReq{TenantId: v.TenantId, Id: v.Id})
						if err != nil {
							e.Log.Errorf("StopSource error:%s \r\n", err)
						}
					}

				}
				//mu.Lock()
				//
				//*partNum++
				//result := float64(*partNum) / float64(workNum)
				//formattedResult := strconv.FormatFloat(result, 'f', 2, 64)
				//param["progress"] = formattedResult
				//if *partNum == workNum {
				//	err := e.Orm.Model(live_schedule.ChannelBasic{}).Where("id=?", channelId).Updates(param).Error
				//	if err != nil {
				//		e.Log.Errorf("Update ChannelTask error:%s \r\n", err)
				//	}
				//} else {
				//	err := e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", channelId).Update("progress", formattedResult).Error
				//	if err != nil {
				//		e.Log.Errorf("ChannelBasic Updates error:%s \r\n", err)
				//	}
				//}
				//e.Log.Trace("当前运行的任务个数：", *partNum, workNum)
				//mu.Unlock()
			}
		}()
	}
	return nil
}

func (e *ChannelTaskService) CheckScheduleCount(tenantId string) int64 {
	var convertCount int64
	err := e.Orm.Model(&live_schedule.ChannelTask{}).Where("tenant_id=? and task_id !='' and area<=1", tenantId).Count(&convertCount).Error
	if err != nil {
		e.Log.Errorf("get ChannelTask error:%s \r\n", err)
	}
	var cortexCount int64
	err = e.Orm.Model(&live_schedule.CortexTask{}).Where("tenant_id=? and task_id !='' and area<=1", tenantId).Count(&cortexCount).Error
	if err != nil {
		e.Log.Errorf("get CortexTask error:%s \r\n", err)
	}
	var transCount int64
	err = e.Orm.Model(&live_schedule.ChannelTransCodeTask{}).Where("tenant_id=? and task_id !='' and area<=1", tenantId).Count(&transCount).Error
	if err != nil {
		e.Log.Errorf("get ChannelTransCodeTask error:%s \r\n", err)
	}
	var auditCount int64
	err = e.Orm.Model(&live_schedule.AuditTask{}).Where("tenant_id=? and task_id !='' and area<=1", tenantId).Count(&auditCount).Error
	if err != nil {
		e.Log.Errorf("get AuditTask error:%s \r\n", err)
	}
	var waterMarkCount int64
	err = e.Orm.Model(&live_schedule.WaterMarkTransTask{}).Where("tenant_id=? and task_id !='' and area<=1", tenantId).Count(&waterMarkCount).Error
	if err != nil {
		e.Log.Errorf("get WaterMarkTransTask error:%s \r\n", err)
	}
	return convertCount + cortexCount + transCount + auditCount + waterMarkCount
}

func (e *ChannelTaskService) StartOrStopTaskByTrafficOn(req *dto.StopOrStopTaskByTrafficOnReq) error {
	cortexTaskService := &CortexTaskService{}
	cortexTaskService.Log = e.Log
	cortexTaskService.Orm = e.Orm
	auditTaskService := &AuditTaskService{}
	auditTaskService.Log = e.Log
	auditTaskService.Orm = e.Orm
	recordTaskService := &RecordTaskService{}
	recordTaskService.Log = e.Log
	recordTaskService.Orm = e.Orm
	convertTaskService := &ConvertTaskService{}
	convertTaskService.Log = e.Log
	convertTaskService.Orm = e.Orm
	transCodeTaskService := &TransCodeTaskService{}
	transCodeTaskService.Log = e.Log
	transCodeTaskService.Orm = e.Orm
	waterMarkTaskService := &WaterMarkTaskService{}
	waterMarkTaskService.Log = e.Log
	waterMarkTaskService.Orm = e.Orm
	var wg sync.WaitGroup

	var partNum int
	convertTasks := make([]live_schedule.ChannelTask, 0, 5)
	err := e.Orm.Model(&live_schedule.ChannelTask{}).Where("traffic_on_id = ?", req.UpId).Find(&convertTasks).Error
	if err != nil {
		e.Log.Errorf(" get ChannelTask error:%s \r\n", err)
		return err
	}

	trafficOn := live_schedule.TrafficOn{}
	err = e.Orm.Model(&live_schedule.TrafficOn{}).Where("id = ?", req.UpId).Find(&trafficOn).Error
	if err != nil {
		e.Log.Errorf(" get TrafficOn error:%s \r\n", err)
		return err
	}
	now := time.Now().Unix()
	if req.Type == 1 && trafficOn.StateUpdateTime+30 > now {
		return errors.New(string(apis.UpStartTimeError.Code))
	}
	basic := live_schedule.ChannelBasic{}
	err = e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id = ?", trafficOn.ChannelId).Find(&basic).Error
	if err != nil {
		e.Log.Errorf(" get ChannelBasic error:%s \r\n", err)
		return err
	}
	if basic.ChannelState == 2 && req.Type == 1 {
		return errors.New(string(apis.UpStartError.Code))
	}

	cortexTasks := make([]live_schedule.CortexTask, 0, 5)
	err = e.Orm.Model(&live_schedule.CortexTask{}).Where("traffic_on_id = ? and task_id!=''", req.UpId).Find(&cortexTasks).Error
	if err != nil {
		e.Log.Errorf(" get cortexTasks error:%s \r\n", err)
		return err
	}
	sourceRecordTasks := make([]live_schedule.SourceRecordTask, 0, 5)
	err = e.Orm.Model(&live_schedule.SourceRecordTask{}).Where("traffic_on_id = ?", req.UpId).Find(&sourceRecordTasks).Error
	if err != nil {
		e.Log.Errorf(" get SourceRecordTask error:%s \r\n", err)
		return err
	}

	auditTasks := make([]live_schedule.AuditTask, 0, 5)
	err = e.Orm.Model(&live_schedule.AuditTask{}).Where("traffic_on_id = ?", req.UpId).Find(&auditTasks).Error
	if err != nil {
		e.Log.Errorf(" get auditTasks error:%s \r\n", err)
		return err
	}

	waterMarkTransTasks := make([]live_schedule.WaterMarkTransTask, 0, 5)
	err = e.Orm.Model(&live_schedule.WaterMarkTransTask{}).Where("traffic_on_id = ?", req.UpId).Find(&waterMarkTransTasks).Error
	if err != nil {
		e.Log.Errorf(" get waterMarkTransTasks error:%s \r\n", err)
		return err
	}

	transCodeTasks := make([]live_schedule.ChannelTransCodeTask, 0, 5)
	err = e.Orm.Model(&live_schedule.ChannelTransCodeTask{}).Where("traffic_on_id = ?", req.UpId).Find(&transCodeTasks).Error
	if err != nil {
		e.Log.Errorf(" get transCodeTasks error:%s \r\n", err)
		return err
	}

	if req.Type == 1 {
		convertTaskService.StartChannelTasks(&wg, convertTasks, 0, &partNum, nil, nil)
		cortexTaskService.StartCortexTasks(&wg, cortexTasks, 0, &partNum, nil, nil)
		recordTaskService.StartRecordTasks(&wg, sourceRecordTasks, 0, &partNum, nil, nil)
		waterMarkTaskService.StartWatermarkTasks(&wg, waterMarkTransTasks, 0, &partNum, nil, nil)
		transCodeTaskService.StartTransCodeTasks(&wg, transCodeTasks, 0, &partNum, nil, nil)
	} else {
		convertTaskService.StopChannelTasks(&wg, convertTasks, 0, &partNum, nil, nil)
		cortexTaskService.StopCortexTasks(&wg, cortexTasks, 0, &partNum, nil, nil)
		recordTaskService.StopRecordTasks(&wg, sourceRecordTasks, 0, &partNum, nil, nil)
		waterMarkTaskService.StopWaterMarkTasks(&wg, waterMarkTransTasks, 0, &partNum, nil, nil)
		transCodeTaskService.StopTransCodeTasks(&wg, transCodeTasks, 0, &partNum, nil, nil)
	}

	param := make(map[string]interface{})
	if req.Type == 2 {
		param["state_update_time"] = now
	}
	param["traffic_on_state"] = req.Type
	err = e.Orm.Model(&live_schedule.TrafficOn{}).Where("id=?", req.UpId).Updates(&param).Error
	if err != nil {
		e.Log.Errorf(" update TrafficOn error:%s \r\n", err)
		return err
	}

	wg.Wait()
	return nil

}

func (e *ChannelTaskService) StartOrStopTaskByTrafficOff(req *dto.StopOrStopTaskByTrafficOffReq) error {
	cortexTaskService := &CortexTaskService{}
	cortexTaskService.Log = e.Log
	cortexTaskService.Orm = e.Orm
	auditTaskService := &AuditTaskService{}
	auditTaskService.Log = e.Log
	auditTaskService.Orm = e.Orm
	recordTaskService := &RecordTaskService{}
	recordTaskService.Log = e.Log
	recordTaskService.Orm = e.Orm
	convertTaskService := &ConvertTaskService{}
	convertTaskService.Log = e.Log
	convertTaskService.Orm = e.Orm
	transCodeTaskService := &TransCodeTaskService{}
	transCodeTaskService.Log = e.Log
	transCodeTaskService.Orm = e.Orm
	waterMarkTaskService := &WaterMarkTaskService{}
	waterMarkTaskService.Log = e.Log
	waterMarkTaskService.Orm = e.Orm
	var wg sync.WaitGroup

	trafficOff := live_schedule.TrafficOff{}
	err := e.Orm.Model(&live_schedule.TrafficOff{}).Where("id = ?", req.OffId).Find(&trafficOff).Error
	if err != nil {
		e.Log.Errorf(" get TrafficOff error:%s \r\n", err)
		return err
	}
	now := time.Now().Unix()
	if req.Type == 1 && trafficOff.StateUpdateTime+30 > now {
		return errors.New(string(apis.UpStartTimeError.Code))
	}

	var partNum int
	convertTasks := make([]live_schedule.ChannelTask, 0, 5)
	err = e.Orm.Model(&live_schedule.ChannelTask{}).Where("traffic_off_id = ?", req.OffId).Find(&convertTasks).Error
	if err != nil {
		e.Log.Errorf(" get ChannelTask error:%s \r\n", err)
		return err
	}

	cortexTasks := make([]live_schedule.CortexTask, 0, 5)
	err = e.Orm.Model(&live_schedule.CortexTask{}).Where("traffic_off_id = ?", req.OffId).Find(&cortexTasks).Error
	if err != nil {
		e.Log.Errorf(" get cortexTasks error:%s \r\n", err)
		return err
	}
	//sourceRecordTasks := make([]live_schedule.SourceRecordTask, 0, 5)
	//err = e.Orm.Model(&live_schedule.SourceRecordTask{}).Where("traffic_off_id = ?", req.OffId).Find(&sourceRecordTasks).Error
	//if err != nil {
	//	e.Log.Errorf(" get SourceRecordTask error:%s \r\n", err)
	//	return err
	//}

	auditTasks := make([]live_schedule.AuditTask, 0, 5)
	err = e.Orm.Model(&live_schedule.AuditTask{}).Where("traffic_off_id = ?", req.OffId).Find(&auditTasks).Error
	if err != nil {
		e.Log.Errorf(" get auditTasks error:%s \r\n", err)
		return err
	}

	waterMarkTransTasks := make([]live_schedule.WaterMarkTransTask, 0, 5)
	err = e.Orm.Model(&live_schedule.WaterMarkTransTask{}).Where("traffic_off_id = ?", req.OffId).Find(&waterMarkTransTasks).Error
	if err != nil {
		e.Log.Errorf(" get waterMarkTransTasks error:%s \r\n", err)
		return err
	}

	transCodeTasks := make([]live_schedule.ChannelTransCodeTask, 0, 5)
	err = e.Orm.Model(&live_schedule.ChannelTransCodeTask{}).Where("traffic_off_id = ?", req.OffId).Find(&transCodeTasks).Error
	if err != nil {
		e.Log.Errorf(" get transCodeTasks error:%s \r\n", err)
		return err
	}

	if req.Type == 1 {
		convertTaskService.StartChannelTasks(&wg, convertTasks, 0, &partNum, nil, nil)
		cortexTaskService.StartCortexTasks(&wg, cortexTasks, 0, &partNum, nil, nil)
		//recordTaskService.StartRecordTasks(sourceRecordTasks, 0, &partNum, nil, nil)
		waterMarkTaskService.StartWatermarkTasks(&wg, waterMarkTransTasks, 0, &partNum, nil, nil)
		transCodeTaskService.StartTransCodeTasks(&wg, transCodeTasks, 0, &partNum, nil, nil)
	} else {
		convertTaskService.StopChannelTasks(&wg, convertTasks, 0, &partNum, nil, nil)
		cortexTaskService.StopCortexTasks(&wg, cortexTasks, 0, &partNum, nil, nil)
		//recordTaskService.StopRecordTasks(sourceRecordTasks, 0, &partNum, nil, nil)
		waterMarkTaskService.StopWaterMarkTasks(&wg, waterMarkTransTasks, 0, &partNum, nil, nil)
		transCodeTaskService.StopTransCodeTasks(&wg, transCodeTasks, 0, &partNum, nil, nil)
	}
	//err = e.Orm.Model(&live_schedule.TrafficOff{}).Where("id=?", req.OffId).Update("traffic_off_state", req.Type).Error
	//if err != nil {
	//	e.Log.Errorf(" update TrafficOff error:%s \r\n", err)
	//	return err
	//}
	param := make(map[string]interface{})
	if req.Type == 2 {
		param["state_update_time"] = now
	}
	param["traffic_off_state"] = req.Type
	err = e.Orm.Model(&live_schedule.TrafficOff{}).Where("id=?", req.OffId).Updates(&param).Error
	if err != nil {
		e.Log.Errorf(" update TrafficOff error:%s \r\n", err)
		return err
	}
	wg.Wait()
	return nil

}

func (e *ChannelTaskService) StartOrStopTaskByMediaProcess(req *dto.StopOrStopTaskByMediaProcessReq) error {
	cortexTaskService := &CortexTaskService{}
	cortexTaskService.Log = e.Log
	cortexTaskService.Orm = e.Orm
	auditTaskService := &AuditTaskService{}
	auditTaskService.Log = e.Log
	auditTaskService.Orm = e.Orm
	recordTaskService := &RecordTaskService{}
	recordTaskService.Log = e.Log
	recordTaskService.Orm = e.Orm
	convertTaskService := &ConvertTaskService{}
	convertTaskService.Log = e.Log
	convertTaskService.Orm = e.Orm
	transCodeTaskService := &TransCodeTaskService{}
	transCodeTaskService.Log = e.Log
	transCodeTaskService.Orm = e.Orm
	waterMarkTaskService := &WaterMarkTaskService{}
	waterMarkTaskService.Log = e.Log
	waterMarkTaskService.Orm = e.Orm
	var wg sync.WaitGroup

	mediaProcess := live_schedule.MediaProcess{}
	err := e.Orm.Model(&live_schedule.MediaProcess{}).Where("id = ?", req.MediaProcessId).Find(&mediaProcess).Error
	if err != nil {
		e.Log.Errorf(" get MediaProcess error:%s \r\n", err)
		return err
	}
	now := time.Now().Unix()
	if req.Type == 1 && mediaProcess.StateUpdateTime+30 > now {
		return errors.New(string(apis.UpStartTimeError.Code))
	}

	var partNum int
	media := live_schedule.MediaProcess{}
	err = e.Orm.Model(&live_schedule.MediaProcess{}).Where("id=?", req.MediaProcessId).Find(&media).Error
	if err != nil {
		e.Log.Errorf(" get MediaProcess error:%s \r\n", err)
		return err
	}
	switch media.MediaType {
	case 1:
		convertTasks := make([]live_schedule.ChannelTask, 0, 5)
		db := e.Orm.Model(&live_schedule.ChannelTask{}).Where("media_id = ?", req.MediaProcessId)
		if req.TrafficOnId > 0 {
			db.Where("traffic_on_id =?", req.TrafficOnId)
		}
		err = db.Find(&convertTasks).Error
		if err != nil {
			e.Log.Errorf(" get ChannelTask error:%s \r\n", err)
			return err
		}
		if req.Type == 1 {
			convertTaskService.StartChannelTasks(&wg, convertTasks, 0, &partNum, nil, nil)
		} else {
			convertTaskService.StopChannelTasks(&wg, convertTasks, 0, &partNum, nil, nil)
		}
	case 2:
		cortexTasks := make([]live_schedule.CortexTask, 0, 5)
		db := e.Orm.Model(&live_schedule.CortexTask{}).Where("media_process_id = ?", req.MediaProcessId)

		if req.TrafficOnId > 0 {
			db.Where("traffic_on_id =?", req.TrafficOnId)
		}
		err = db.Find(&cortexTasks).Error
		if err != nil {
			e.Log.Errorf(" get cortexTasks error:%s \r\n", err)
			return err
		}
		if req.Type == 1 {
			cortexTaskService.StartCortexTasks(&wg, cortexTasks, 0, &partNum, nil, nil)
		} else {
			cortexTaskService.StopCortexTasks(&wg, cortexTasks, 0, &partNum, nil, nil)
		}
	case 3:
		auditTasks := make([]live_schedule.AuditTask, 0, 5)
		db := e.Orm.Model(&live_schedule.AuditTask{}).Where("media_id = ?", req.MediaProcessId)
		if err != nil {
			e.Log.Errorf(" get auditTasks error:%s \r\n", err)
			return err
		}
		if req.TrafficOnId > 0 {
			db.Where("traffic_on_id =?", req.TrafficOnId)
		}
		err = db.Find(&auditTasks).Error
		if err != nil {
			e.Log.Errorf(" get auditTasks error:%s \r\n", err)
			return err
		}
		//if req.Type == 1 {
		//	auditTaskService.StartAuditTasks(auditTasks, media, 0, &partNum, nil, nil)
		//} else {
		//	auditTaskService.StopAuditTasks(auditTasks, media, 0, &partNum, nil, nil)
		//}
	case 4:
		transCodeTasks := make([]live_schedule.ChannelTransCodeTask, 0, 5)
		db := e.Orm.Model(&live_schedule.ChannelTransCodeTask{}).Where("media_id = ?", req.MediaProcessId)
		if err != nil {
			e.Log.Errorf(" get transCodeTasks error:%s \r\n", err)
			return err
		}
		if req.TrafficOnId > 0 {
			db.Where("traffic_on_id =?", req.TrafficOnId)
		}
		err = db.Find(&transCodeTasks).Error
		if err != nil {
			e.Log.Errorf(" get ChannelTransCodeTask error:%s \r\n", err)
			return err
		}
		if req.Type == 1 {
			transCodeTaskService.StartTransCodeTasks(&wg, transCodeTasks, 0, &partNum, nil, nil)
		} else {
			transCodeTaskService.StopTransCodeTasks(&wg, transCodeTasks, 0, &partNum, nil, nil)
		}

	case 5:
		sourceRecordTasks := make([]live_schedule.SourceRecordTask, 0, 5)
		db := e.Orm.Model(&live_schedule.SourceRecordTask{}).Where("media_id = ?", req.MediaProcessId)
		if err != nil {
			e.Log.Errorf(" get SourceRecordTask error:%s \r\n", err)
			return err
		}
		if req.TrafficOnId > 0 {
			db.Where("traffic_on_id =?", req.TrafficOnId)
		}
		err = db.Find(&sourceRecordTasks).Error
		if err != nil {
			e.Log.Errorf(" get SourceRecordTask error:%s \r\n", err)
			return err
		}
		if req.Type == 1 {
			recordTaskService.StartRecordTasks(&wg, sourceRecordTasks, 0, &partNum, nil, nil)
		} else {
			recordTaskService.StopRecordTasks(&wg, sourceRecordTasks, 0, &partNum, nil, nil)
		}
	case 6:
		waterMarkTransTasks := make([]live_schedule.WaterMarkTransTask, 0, 5)
		db := e.Orm.Model(&live_schedule.WaterMarkTransTask{}).Where("media_id = ?", req.MediaProcessId)
		if err != nil {
			e.Log.Errorf(" get waterMarkTransTasks error:%s \r\n", err)
			return err
		}
		if req.TrafficOnId > 0 {
			db.Where("traffic_on_id =?", req.TrafficOnId)
		}
		err = db.Find(&waterMarkTransTasks).Error
		if err != nil {
			e.Log.Errorf(" get waterMarkTransTasks error:%s \r\n", err)
			return err
		}
		if req.Type == 1 {
			waterMarkTaskService.StartWatermarkTasks(&wg, waterMarkTransTasks, 0, &partNum, nil, nil)
		} else {
			waterMarkTaskService.StopWaterMarkTasks(&wg, waterMarkTransTasks, 0, &partNum, nil, nil)
		}

	}
	//state := 3
	//if req.TrafficOnId <= 0 {
	//	state = req.Type
	//} else {
	//	states := make([]int, 0, 5)
	//	switch media.MediaType {
	//	case 1:
	//		err = e.Orm.Model(&live_schedule.ChannelTask{}).Select("task_state").Where("media_id = ?", req.MediaProcessId).Find(&states).Error
	//		if err != nil {
	//			e.Log.Errorf(" get ChannelTask error:%s \r\n", err)
	//
	//		}
	//	case 2:
	//		err = e.Orm.Model(&live_schedule.CortexTask{}).Select("task_state").Where("media_id = ? and task_state!=''", req.MediaProcessId).Find(&states).Error
	//		if err != nil {
	//			e.Log.Errorf(" get CortexTask error:%s \r\n", err)
	//
	//		}
	//		fmt.Println("kkkkkkkk:", states)
	//	case 3:
	//		err = e.Orm.Model(&live_schedule.AuditTask{}).Select("task_state").Where("media_id = ?", req.MediaProcessId).Find(&states).Error
	//		if err != nil {
	//			e.Log.Errorf(" get AuditTask error:%s \r\n", err)
	//
	//		}
	//	case 4:
	//		err = e.Orm.Model(&live_schedule.ChannelTransCodeTask{}).Select("task_state").Where("media_id = ?", req.MediaProcessId).Find(&states).Error
	//		if err != nil {
	//			e.Log.Errorf(" get ChannelTransCodeTask error:%s \r\n", err)
	//
	//		}
	//	case 5:
	//		err = e.Orm.Model(&live_schedule.SourceRecordTask{}).Select("task_state").Where("media_id = ?", req.MediaProcessId).Find(&states).Error
	//		if err != nil {
	//			e.Log.Errorf(" get SourceRecordTask error:%s \r\n", err)
	//
	//		}
	//	case 6:
	//		err = e.Orm.Model(&live_schedule.WaterMarkTransTask{}).Select("task_state").Where("media_id = ?", req.MediaProcessId).Find(&states).Error
	//		if err != nil {
	//			e.Log.Errorf(" get WaterMarkTransTask error:%s \r\n", err)
	//
	//		}
	//	}
	//
	//	if allSame(states, 1) {
	//		state = 1
	//	} else if allSame(states, 2) {
	//		state = 2
	//	} else if allSame(states, 3) {
	//		state = 3
	//	} else {
	//		state = 4
	//	}
	//}
	param := make(map[string]interface{})
	if req.Type == 2 {
		param["state_update_time"] = now
	}
	param["media_state"] = req.Type
	err = e.Orm.Model(&live_schedule.MediaProcess{}).Where("id=?", req.MediaProcessId).Updates(&param).Error
	if err != nil {
		e.Log.Errorf(" update MediaProcess error:%s \r\n", err)
		return err
	}
	wg.Wait()
	return nil

}

func allSame(slice []int, target int) bool {
	for _, num := range slice {
		if num != target {
			return false
		}
	}
	return true
}

func (e *ChannelTaskService) GetUpMedia(req *dto.GetUpMediaReq) ([]live_schedule.UpMedia, error) {

	upMedias := make([]live_schedule.UpMedia, 0, 5)
	media := live_schedule.MediaProcess{}
	err := e.Orm.Model(&live_schedule.MediaProcess{}).Where("id=?", req.MediaProcessId).Find(&media).Error
	if err != nil {
		e.Log.Errorf("get MediaProcess error:%s \r\n", err)
		return upMedias, err
	}
	//live_schedule.ChannelTransCodeTask{}
	db := e.Orm.Model(&live_schedule.UpMedia{})
	sqlStr := "up_media.media_id ,traffic_on.traffic_on_name,up_media.up_id as up_id,source_basic.source_name,source_basic.id as source_id,"
	switch media.MediaType {
	case 1:
		db.Select(sqlStr + "channel_task.task_state as task_state").
			Joins("left JOIN traffic_on ON up_media.up_id = traffic_on.id").
			Joins("left JOIN channel_task ON channel_task.traffic_on_id = traffic_on.id")
	case 2:
		db.Select(sqlStr + "cortex_task.task_state as task_state").
			Joins("left JOIN traffic_on ON up_media.up_id = traffic_on.id").
			Joins("left JOIN cortex_task ON cortex_task.traffic_on_id = traffic_on.id and cortex_task.task_id !=''")
	case 3:
		db.Select(sqlStr + "audit_task.task_state as task_state").
			Joins("left JOIN traffic_on ON up_media.up_id = traffic_on.id").
			Joins("left JOIN audit_task ON audit_task.traffic_on_id = traffic_on.id")
	case 4:
		db.Select(sqlStr + "channel_trans_code_task.task_state as task_state").
			Joins("left JOIN traffic_on ON up_media.up_id = traffic_on.id").
			Joins("left JOIN channel_trans_code_task ON channel_trans_code_task.traffic_on_id = traffic_on.id")
	case 5:
		db.Select(sqlStr + "record_task.task_state as task_state").
			Joins("left JOIN traffic_on ON up_media.up_id = traffic_on.id").
			Joins("left JOIN record_task ON record_task.traffic_on_id = traffic_on.id")
	case 6:
		db.Select(sqlStr + "watermark_trans_task.task_state as task_state").
			Joins("left JOIN traffic_on ON up_media.up_id = traffic_on.id").
			Joins("left JOIN watermark_trans_task ON watermark_trans_task.traffic_on_id = traffic_on.id")
	default:

	}

	err = db.
		Joins("left JOIN source_basic ON traffic_on.source_id = source_basic.id").Where("up_media.media_id=?", req.MediaProcessId).
		Find(&upMedias).
		Error
	if err != nil {
		e.Log.Errorf("get UpMedia error:%s \r\n", err)
		return upMedias, err
	}

	upTaskMap := make(map[int][]live_schedule.UpMedia)
	for _, v := range upMedias {
		v := v
		fmt.Println(v)
		medias, ok := upTaskMap[v.UpId]
		if !ok {
			medias = make([]live_schedule.UpMedia, 0, 5)
		}
		medias = append(medias, v)
		upTaskMap[v.UpId] = medias
	}

	showMedias := make([]live_schedule.UpMedia, 0, 3)
	for k, tasks := range upTaskMap {
		k := k
		media := live_schedule.UpMedia{}
		media.UpId = k
		taskState := 3
		for _, task := range tasks {
			if task.TaskState == 2 {
				taskState = 2
				break
			}
			if task.TaskState == 3 {
				taskState = 3
			}
			if task.TaskState == 1 {
				taskState = 1
			}
			media.TrafficOnName = task.TrafficOnName
			media.MediaId = task.MediaId
			media.SourceId = task.SourceId
			media.SourceName = task.SourceName
		}
		media.TaskState = taskState
		showMedias = append(showMedias, media)
	}
	return showMedias, nil

}

func (e *ChannelTaskService) SaveWorkflow1(c *dto.SaveWorkflowReq, basic live_schedule.ChannelBasic) error {

	liveService := &liveService.LiveBasic{}
	streamHubService := &streamHubService.StreamHubService{}
	baiduService := &baiduService.BaiduService{}
	streamHubService.Log = e.Log
	liveService.Log = e.Log
	baiduService.Log = e.Log
	cortexTaskService := &CortexTaskService{}
	cortexTaskService.Log = e.Log
	cortexTaskService.Orm = e.Orm

	liveBasicService := SourceBasicService{}
	liveBasicService.Log = e.Log
	liveBasicService.Orm = e.Orm

	auditTaskService := AuditTaskService{}
	auditTaskService.Log = e.Log
	auditTaskService.Orm = e.Orm

	recordTaskService := RecordTaskService{}
	recordTaskService.Log = e.Log
	recordTaskService.Orm = e.Orm

	convertTaskService := &ConvertTaskService{}
	convertTaskService.Log = e.Log
	convertTaskService.Orm = e.Orm

	transCodeTaskService := &TransCodeTaskService{}
	transCodeTaskService.Log = e.Log
	transCodeTaskService.Orm = e.Orm

	waterMarkTaskService := WaterMarkTaskService{}
	waterMarkTaskService.Log = e.Log
	waterMarkTaskService.Orm = e.Orm
	if len(c.NewChartData) > 0 {
		err := e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", c.ChannelId).Update("new_chart_data", c.NewChartData).Error
		if err != nil {
			e.Log.Errorf("Update ChannelBasic error:%s \r\n", err)
			return err
		}
		err = json.Unmarshal([]byte(c.NewChartData), &c.ChartData)
		if err != nil {
			e.Log.Errorf("Unmarshal NewChartData error:%s \r\n", err)
			return err
		}
	}

	portUrlMap := make(map[string]string)
	nodeMap := make(map[string]live_schedule.NodeData)
	inputNodes := make([]live_schedule.NodeData, 0, 5)
	for _, nodeData := range c.ChartData.NodeData {
		nodeData := nodeData
		for _, outUrl := range nodeData.Node.Data.OutUrls {
			portUrlMap[outUrl.PortId] = outUrl.OutputUrl
		}
		nodeMap[nodeData.Node.Id] = nodeData
		if nodeData.Node.Data.Type == "input" {
			inputNodes = append(inputNodes, nodeData)
		}
	}

	oldData := live_schedule.ChartData{}
	if len(basic.ChartData) > 0 {
		err := json.Unmarshal([]byte(basic.ChartData), &oldData)
		if err != nil {
			e.Log.Errorf("Unmarshal ChartData error:%s \r\n", err)
			return err
		}
	}

	if len(basic.ChartData) > 0 {
		//判断是否需要删除节点任务
		err := e.DeleteNode(oldData, c.ChartData, c.ChannelId, c.TenantId, streamHubService)
		if err != nil {
			e.Log.Errorf("DeleteNode error:%s \r\n", err)
			return err
		}
	}

	//for _, node := range c.ChartData.NodeData {
	//	node := node
	//	//treeNode := live_schedule.TreeNode{}
	//	//treeNode.ParaNode = node
	//	inputModels := make([]live_schedule.InputModel, 0, 5)
	//	for _, edgeData := range c.ChartData.EdgeData {
	//		if edgeData.Target.Id == node.Node.Id {
	//			inputModel := live_schedule.InputModel{}
	//			inputModel.InputId = edgeData.Source.Id
	//			inputModel.InputPort = edgeData.Source.Port
	//			inputModel.SelfPort = edgeData.Target.Port
	//			inputModel.SelfId = edgeData.Target.Id
	//			inputModel.InputUrl = portUrlMap[edgeData.Source.Port]
	//			inputModels = append(inputModels, inputModel)
	//		}
	//	}
	//}

	err := e.AddNodeInfo(inputNodes, &c.ChartData, nodeMap, portUrlMap, basic, streamHubService, transCodeTaskService, convertTaskService, &oldData)
	if err != nil {
		e.Log.Errorf("AddNodeInfo error:%s \r\n", err)
		return err
	}

	err = e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id = ?", c.ChannelId).Find(&basic).Error
	if err != nil {
		e.Log.Errorf("get ChannelBasic error:%s \r\n", err)
		return err
	}
	if len(basic.ChartData) > 0 {

		err = e.UpdateNode(oldData, c.ChartData, basic, portUrlMap, streamHubService, transCodeTaskService, convertTaskService)
		if err != nil {
			e.Log.Errorf("UpdateNode error:%s \r\n", err)
			return err
		}
	}

	////新增节点任务
	//for k, nodeData := range c.ChartData.NodeData {
	//	//k := 0
	//	node := nodeMap[nodeData.Node.Id]
	//	if node.Node.Data.Type == "input" {
	//		continue
	//	}
	//	inputModels := make([]live_schedule.InputModel, 0, 5)
	//	for _, edgeData := range c.ChartData.EdgeData {
	//		if edgeData.Target.Id == node.Node.Id {
	//			inputModel := live_schedule.InputModel{}
	//			inputModel.InputId = edgeData.Source.Id
	//			inputModel.InputPort = edgeData.Source.Port
	//			inputModel.SelfPort = edgeData.Target.Port
	//			inputModel.SelfId = edgeData.Target.Id
	//			inputModel.InputUrl = portUrlMap[edgeData.Source.Port]
	//			inputModels = append(inputModels, inputModel)
	//		}
	//	}
	//
	//	e.Log.Trace("输入信息为：", inputModels)
	//	switch node.Node.Data.Type {
	//	case "media_switch":
	//		switchTask := live_schedule.SwitchTask{}
	//		err := e.Orm.Model(&live_schedule.SwitchTask{}).Where("channel_id=? and media_id=?", c.ChannelId, node.Node.Id).Find(&switchTask).Error
	//		if err != nil {
	//			e.Log.Errorf("get SwitchTask error:%s \r\n", err)
	//			return err
	//		}
	//		switchTask.TenantId = c.TenantId
	//		switchTask.ChannelId = c.ChannelId
	//		switchUrl, err := e.MediaSwitch(switchTask, inputModels, node.Node, streamHubService)
	//		if err != nil {
	//			e.Log.Errorf("MediaSwitch error:%s \r\n", err)
	//			return err
	//		}
	//		*(&c.ChartData.NodeData[k].Node.Data.OutUrls[0].OutputUrl) = switchUrl
	//		for _, nodeData := range c.ChartData.NodeData {
	//			nodeData := nodeData
	//			for _, outUrl := range nodeData.Node.Data.OutUrls {
	//				portUrlMap[outUrl.PortId] = outUrl.OutputUrl
	//			}
	//		}
	//	case "media_trans":
	//		channelTransTask := live_schedule.ChannelTransTask{}
	//		err := e.Orm.Model(&live_schedule.ChannelTransTask{}).Where("channel_id=? and media_id=?", c.ChannelId, node.Node.Id).Find(&channelTransTask).Error
	//		if err != nil {
	//			e.Log.Errorf("get ChannelTransTask error:%s \r\n", err)
	//			return err
	//		}
	//		channelTransTask.TenantId = c.TenantId
	//		channelTransTask.ChannelId = c.ChannelId
	//		transUrl, err := transCodeTaskService.CreateTransCodeTask(node.Node, &channelTransTask, inputModels)
	//		if err != nil {
	//			e.Log.Errorf("CreateTransCodeTask error:%s \r\n", err)
	//			return err
	//		}
	//		*(&c.ChartData.NodeData[k].Node.Data.OutUrls[0].OutputUrl) = transUrl
	//		for _, nodeData := range c.ChartData.NodeData {
	//			nodeData := nodeData
	//			for _, outUrl := range nodeData.Node.Data.OutUrls {
	//				portUrlMap[outUrl.PortId] = outUrl.OutputUrl
	//			}
	//		}
	//
	//	case "output":
	//		channelTask := live_schedule.CovertTask{}
	//		err := e.Orm.Model(&live_schedule.CovertTask{}).Where("channel_id=? and media_id=?", c.ChannelId, node.Node.Id).Find(&channelTask).Error
	//		if err != nil {
	//			e.Log.Errorf("get ChannelTask error:%s \r\n", err)
	//			return err
	//		}
	//		channelTask.ChannelId = c.ChannelId
	//		channelTask.TenantId = c.TenantId
	//		err = convertTaskService.CreateChannelTask1(node.Node, &channelTask, inputModels)
	//		if err != nil {
	//			e.Log.Errorf("CreateChannelTask1 error:%s \r\n", err)
	//			return err
	//		}
	//	case "":
	//	default:
	//
	//	}
	//
	//}

	return nil
}

func (e *ChannelTaskService) MediaSwitch(task *live_schedule.SwitchTask, inputModels []live_schedule.InputModel, node live_schedule.Node, streamHubService *streamHubService.StreamHubService, basic live_schedule.ChannelBasic) (live_schedule.SwitchTask, error) {

	e.Log.Trace("进入切换模式")
	switchReq := &StreamHubDto.SwitchSourceReq{}
	switchReq.Name = ""
	switchReq.TenantId = task.TenantId
	switchReq.Id = task.TaskId
	switchInput := StreamHubDto.SwitchInput{}
	switchInput.Type = "push"
	switchInput.Format = "udp"
	switchInput.Active = 0
	switchInput.Policy = StreamHubDto.Policy{Mode: "manual"}
	switchInput.Sources = make([]StreamHubDto.Source, 0, 3)
	switchReq.Type = "live"
	if node.Data.IsCustom == 1 {
		switchReq.OutputAddress = node.Data.CustomUrl
	}
	if node.Data.SwitchMode == 1 {
		switchReq.AutoSwitch = 1
	} else {
		switchReq.AutoSwitch = 0
	}

	for _, v := range inputModels {
		if len(v.InputUrl) <= 0 {
			continue
		}
		switchSource := StreamHubDto.Source{}
		switchSource.Url = v.InputUrl
		switchInput.Sources = append(switchInput.Sources, switchSource)
	}
	//switchUrl := ""
	switchReq.SwitchInput = switchInput
	switchTask := live_schedule.SwitchTask{}
	if len(task.TaskId) <= 0 {
		switchRes, err := streamHubService.CreateSwitchSourceTask(switchReq)
		if err != nil {
			e.Log.Errorf("CreateSwitchSourceTask error:%s \r\n", err)
			return switchTask, err
		}

		switchTask.TenantId = task.TenantId
		switchTask.ChannelId = task.ChannelId
		switchTask.MediaId = node.Id
		switchTask.ActiveItem = 0
		if switchRes != nil {
			task.TaskId = switchRes.TaskId
			switchTask.TaskId = switchRes.TaskId
			switchTask.OutUrl = switchRes.OutputAddress
			node.Data.OutUrls[0].OutputUrl = switchRes.OutputAddress
			err = e.Orm.Save(&switchTask).Error
			if err != nil {
				e.Log.Errorf("save switchTask error:%s \r\n", err)
				return switchTask, err
			}
			//switchUrl = switchRes.OutputAddress
		}
	} else {
		_, err := streamHubService.UpdateSwitchSourceTask(switchReq)
		if err != nil {
			e.Log.Errorf("CreateSwitchSourceTask error:%s \r\n", err)
			return switchTask, err
		}
	}
	if basic.ChannelState != 2 {
		err := streamHubService.StartSwitchSource(&StreamHubDto.StartOrStopSwitchSourceReq{TenantId: basic.TenantId, Id: task.TaskId})
		if err != nil {
			e.Log.Errorf("StartSwitchSource error:%s \r\n", err)
			return *task, err
		}
	}

	return switchTask, nil

}

func (e *ChannelTaskService) AutoSwitchTask(c *dto.AutoSwitchReq) error {
	if len(c.SwitchId) <= 0 {
		return nil
	}
	param := make(map[string]interface{})
	active := c.ActiveIndex + 1
	param["active_item"] = active
	param["update_time"] = time.Now().Unix()
	err := e.Orm.Model(&live_schedule.SourceBasic{}).Where("origin_switch_task_id=?", c.SwitchId).Updates(&param).Error
	if err != nil {
		e.Log.Errorf("update SourceDetail error:%s", err)
		return err
	}
	return nil
}

func (e *ChannelTaskService) TransCallBack(d dto.CallBackData) error {

	param := make(map[string]interface{})
	param["task_state"] = d.State
	param["update_time"] = time.Now().Unix()
	e.Log.Trace("任务id:", d.TaskId, "任务状态：", d.State)
	err := e.Orm.Model(&live_schedule.ChannelTransTask{}).Where("task_id=?", d.TaskId).Updates(&param).Error
	if err != nil {
		e.Log.Errorf("update ChannelTransTask error:%s \r\n", err)
		return err
	}
	return nil
}

func (e *ChannelTaskService) AddNodeInfo(inputNodes []live_schedule.NodeData, chartData *live_schedule.ChartData, nodeMap map[string]live_schedule.NodeData, portUrlMap map[string]string, basic live_schedule.ChannelBasic, streamHubService *streamHubService.StreamHubService, transCodeTaskService *TransCodeTaskService, convertTaskService *ConvertTaskService, oldChartData *live_schedule.ChartData) error {

	e.Log.Trace("进入添加节点")
	if len(inputNodes) <= 0 {
		return nil
	}
	//portUrlMap := make(map[string]string)
	newInputNodes := make([]live_schedule.NodeData, 0, 5)
	processedMap := make(map[string]struct{})
	isChange := false
	for _, node := range inputNodes {
		node := node
		//treeNode := live_schedule.TreeNode{}
		//treeNode.ParaNode = node
		for _, item := range node.Node.Ports.Items {
			item := item
			for _, edgeData := range chartData.EdgeData {
				edgeData := edgeData
				//寻找该节点的链接的下一个节点
				if item.Id == edgeData.Source.Port {
					node, ok := nodeMap[edgeData.Target.Id]
					if !ok {
						e.Log.Trace("未找到节点信息")
					}
					//如果已经存在该节点跳过
					_, ok = processedMap[node.Node.Id]
					if ok {
						continue
					}
					newInputNodes = append(newInputNodes, node)

					e.Log.Trace("端口输出映射：", portUrlMap, node.Node.Id)
					inputModels := make([]live_schedule.InputModel, 0, 5)
					for _, edgeData := range chartData.EdgeData {
						if edgeData.Target.Id == node.Node.Id {
							inputModel := live_schedule.InputModel{}
							inputModel.InputId = edgeData.Source.Id
							inputModel.InputPort = edgeData.Source.Port
							inputModel.SelfPort = edgeData.Target.Port
							inputModel.SelfId = edgeData.Target.Id
							inputModel.InputUrl = portUrlMap[edgeData.Source.Port]
							inputModels = append(inputModels, inputModel)
						}
					}
					e.Log.Trace("输入的节点信息为：", inputModels)

					switch node.Node.Data.Type {
					case "input":
						if node.Node.Data.SourceFrom == "2" {

						}

					case "media_switch":
						switchTask := live_schedule.SwitchTask{}
						err := e.Orm.Model(&live_schedule.SwitchTask{}).Where("channel_id=? and media_id=?", basic.Id, node.Node.Id).Find(&switchTask).Error
						if err != nil {
							e.Log.Errorf("get SwitchTask error:%s \r\n", err)
							return err
						}
						if switchTask.Id <= 0 {
							switchTask.TenantId = basic.TenantId
							switchTask.ChannelId = basic.Id
							switchUrl := ""

							if node.Node.Data.SwitchSystem != 2 {
								task, err := e.MediaSwitch(&switchTask, inputModels, node.Node, streamHubService, basic)
								if err != nil {
									e.Log.Errorf("MediaSwitch error:%s \r\n", err)
									return err
								}
								e.Log.Trace("MediaSwitch:", task)
								switchUrl = task.OutUrl
								for k, nodeData := range chartData.NodeData {
									if nodeData.Node.Id == node.Node.Id {
										*(&chartData.NodeData[k].Node.Data.TaskId) = task.TaskId
									}
								}
							} else {
								switchUrl = node.Node.Data.CustomUrl
							}

							for k, nodeData := range chartData.NodeData {
								if nodeData.Node.Id == node.Node.Id {
									*(&chartData.NodeData[k].Node.Data.OutUrls[0].OutputUrl) = switchUrl
									if node.Node.Data.SwitchSystem == 2 {
										*(&chartData.NodeData[k].Node.Data.TaskId) = "third"
										*(&chartData.NodeData[k].Node.Data.NodeState) = 2
									}
								}
							}
							//for k, nodeData := range oldChartData.NodeData {
							//	if nodeData.Node.Id == node.Node.Id {
							//		*(&oldChartData.NodeData[k].Node.Data.OutUrls[0].OutputUrl) = switchUrl
							//	}
							//}

							for _, nodeData := range chartData.NodeData {
								nodeData := nodeData
								for _, outUrl := range nodeData.Node.Data.OutUrls {
									portUrlMap[outUrl.PortId] = outUrl.OutputUrl
								}
							}
							isChange = true
						}

					case "media_trans":
						channelTransTask := live_schedule.ChannelTransTask{}
						err := e.Orm.Model(&live_schedule.ChannelTransTask{}).Where("channel_id=? and media_id=?", basic.Id, node.Node.Id).Find(&channelTransTask).Error
						if err != nil {
							e.Log.Errorf("get ChannelTransTask error:%s \r\n", err)
							return err
						}
						if channelTransTask.Id <= 0 {
							channelTransTask.TenantId = basic.TenantId
							channelTransTask.ChannelId = basic.Id
							channelTransTask.MediaId = node.Node.Id
							transCodeTask, err := transCodeTaskService.CreateTransCodeTask(node.Node, &channelTransTask, inputModels, basic)
							if err != nil {
								e.Log.Errorf("CreateTransCodeTask error:%s \r\n", err)
								return err
							}
							for k, nodeData := range chartData.NodeData {
								if nodeData.Node.Id == node.Node.Id {
									*(&chartData.NodeData[k].Node.Data.OutUrls[0].OutputUrl) = transCodeTask.OutUrl
									*(&chartData.NodeData[k].Node.Data.TaskId) = transCodeTask.TaskId
								}
							}
							//for k, nodeData := range oldChartData.NodeData {
							//	if nodeData.Node.Id == node.Node.Id {
							//		*(&oldChartData.NodeData[k].Node.Data.OutUrls[0].OutputUrl) = transCodeTask.OutUrl
							//	}
							//}
							for _, nodeData := range chartData.NodeData {
								nodeData := nodeData
								for _, outUrl := range nodeData.Node.Data.OutUrls {
									portUrlMap[outUrl.PortId] = outUrl.OutputUrl
								}
							}
							e.Log.Trace("更新后的portUrlMap：", portUrlMap)
							isChange = true
						}

					case "output":
						channelTask := live_schedule.CovertTask{}
						err := e.Orm.Model(&live_schedule.CovertTask{}).Where("channel_id=? and media_id=?", basic.Id, node.Node.Id).Find(&channelTask).Error
						if err != nil {
							e.Log.Errorf("get ChannelTask error:%s \r\n", err)
							return err
						}
						if channelTask.Id <= 0 {
							channelTask.ChannelId = basic.Id
							channelTask.TenantId = basic.TenantId
							convertTask, err := convertTaskService.CreateChannelTask1(node.Node, &channelTask, inputModels, basic)
							if err != nil {
								e.Log.Errorf("CreateChannelTask1 error:%s \r\n", err)
								return err
							}
							for k, nodeData := range chartData.NodeData {
								if nodeData.Node.Id == node.Node.Id {
									*(&chartData.NodeData[k].Node.Data.TaskId) = convertTask.TaskId
									*(&chartData.NodeData[k].Node.Data.SourceUrl) = convertTask.OutUrl
								}
							}
							//for k, nodeData := range oldChartData.NodeData {
							//	if nodeData.Node.Id == node.Node.Id {
							//		*(&oldChartData.NodeData[k].Node.Data.OutUrls[0].OutputUrl) = convertTask.OutUrl
							//	}
							//}
							isChange = true
						}
					default:

					}
					processedMap[node.Node.Id] = struct{}{}
				}
			}
		}
	}
	if isChange {
		data, err := json.Marshal(&chartData)
		if err != nil {
			e.Log.Errorf("Marshal chartData error:%s \r\n", err)
			return err
		}
		err = e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", basic.Id).Update("chart_data", string(data)).Error
		if err != nil {
			e.Log.Errorf("Update ChannelBasic error:%s \r\n", err)
			return err
		}
		for _, nodeData := range chartData.NodeData {
			nodeData := nodeData
			for _, outUrl := range nodeData.Node.Data.OutUrls {
				portUrlMap[outUrl.PortId] = outUrl.OutputUrl
			}
		}

		if len(portUrlMap) > 0 {
			portUrlInfo, _ := json.Marshal(&portUrlMap)
			err := e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", basic.Id).Update("output_urls", string(portUrlInfo)).Error
			if err != nil {
				e.Log.Errorf("Update ChannelBasic error:%s \r\n", err)
				return err
			}
		}
	}

	e.AddNodeInfo(newInputNodes, chartData, nodeMap, portUrlMap, basic, streamHubService, transCodeTaskService, convertTaskService, oldChartData)
	return nil
}

func (e *ChannelTaskService) DeleteNode(oldChartData live_schedule.ChartData, newChartData live_schedule.ChartData, channelId int, tenantId string, streamHubService *streamHubService.StreamHubService) error {
	e.Log.Trace("进入删除节点")
	noExists := make([]live_schedule.NodeData, 0, 5)
	oldDataMap := make(map[string]live_schedule.NodeData)
	exsistMap := make(map[string]struct{})
	sourceBasicService := SourceBasicService{}
	sourceBasicService.Log = e.Log
	sourceBasicService.Orm = e.Orm
	for _, oldData := range oldChartData.NodeData {
		oldData := oldData
		oldDataMap[oldData.Node.Id] = oldData
		isExist := false
		for _, newData := range newChartData.NodeData {
			if oldData.Node.Id == newData.Node.Id {
				isExist = true
				break
			}
		}
		if !isExist {
			exsistMap[oldData.Node.Id] = struct{}{}
			//oldChartData.NodeData = append(oldChartData.NodeData[:k], oldChartData.NodeData[k+1:]...)
			noExists = append(noExists, oldData)
		}
	}
	for _, oldData := range oldChartData.EdgeData {
		oldData := oldData
		isExist := false
		old := oldDataMap[oldData.Target.Id]
		_, ok := exsistMap[oldData.Target.Id]
		if ok {
			continue
		}
		//如果新老节点连线一样跳过
		for _, newData := range newChartData.EdgeData {
			if oldData.Target.Id == newData.Target.Id {
				isExist = true
				break
			}
		}
		if !isExist {
			//oldChartData.NodeData = append(oldChartData.NodeData[:k], oldChartData.NodeData[k+1:]...)
			noExists = append(noExists, old)
		}
	}

	for _, deleteData := range noExists {
		switch deleteData.Node.Data.Type {
		case "input":
			if deleteData.Node.Data.FormData.SourceFrom == "2" {
				err := sourceBasicService.Remove(&dto.SourceBasicDeleteReq{Ids: []int{deleteData.Node.Data.FormData.SourceId}})
				if err != nil {
					e.Log.Errorf("Remove SourceBasic error:%s \r\n", err)
					return err
				}
			}
		case "media_switch":
			switchTask := live_schedule.SwitchTask{}
			err := e.Orm.Model(&live_schedule.SwitchTask{}).Where("channel_id=? and media_id=?", channelId, deleteData.Node.Id).Find(&switchTask).Error
			if err != nil {
				e.Log.Errorf("get SwitchTask error:%s \r\n", err)
				return err
			}
			if len(switchTask.TaskId) > 0 {
				err = streamHubService.StopSwitchSource(&StreamHubDto.StartOrStopSwitchSourceReq{TenantId: tenantId, Id: switchTask.TaskId})
				if err != nil {
					e.Log.Errorf("StopSwitchSource error:%s \r\n", err)
					return err
				}
				err = streamHubService.DeleteSwitchSourceTask(&StreamHubDto.SwitchSourceReq{TenantId: tenantId, Id: switchTask.TaskId})
				if err != nil {
					e.Log.Errorf("DeleteSwitchSourceTask error:%s \r\n", err)
					return err
				}
			}

			err = e.Orm.Model(&live_schedule.SwitchTask{}).Where("channel_id=? and media_id=?", channelId, deleteData.Node.Id).Delete(&switchTask).Error
			if err != nil {
				e.Log.Errorf("delete SwitchTask error:%s \r\n", err)
				return err
			}
		case "media_trans":
			channelTransTask := live_schedule.ChannelTransTask{}
			err := e.Orm.Model(&live_schedule.ChannelTransTask{}).Where("channel_id=? and media_id=?", channelId, deleteData.Node.Id).Find(&channelTransTask).Error
			if err != nil {
				e.Log.Errorf("get ChannelTransTask error:%s \r\n", err)
				return err
			}
			if len(channelTransTask.TaskId) > 0 {
				err = streamHubService.RemoveTransCodeTask(&StreamHubDto.TransCodeReq{TenantId: tenantId, Id: channelTransTask.TaskId})
				if err != nil {
					e.Log.Errorf("RemoveTransCodeTask error:%s \r\n", err)
					return err
				}
			}

			err = e.Orm.Model(&live_schedule.ChannelTransTask{}).Where("channel_id=? and media_id=?", channelId, deleteData.Node.Id).Delete(&channelTransTask).Error
			if err != nil {
				e.Log.Errorf("delete ChannelTransTask error:%s \r\n", err)
				return err
			}
		case "output":
			channelTask := live_schedule.CovertTask{}
			liveService := liveService.LiveBasic{}
			liveService.Orm = e.Orm
			liveService.Log = e.Log
			err := e.Orm.Model(&live_schedule.CovertTask{}).Where("channel_id=? and media_id=?", channelId, deleteData.Node.Id).Find(&channelTask).Error
			if err != nil {
				e.Log.Errorf("get ChannelTask error:%s \r\n", err)
				return err
			}
			if len(channelTask.LiveId) > 0 {
				err = liveService.RemoveLive(&liveDto.StopLiveReq{TenantId: tenantId, LiveId: channelTask.LiveId})
				if err != nil {
					e.Log.Errorf("RemoveLive error:%s \r\n", err)
				}
			}
			if len(channelTask.TaskId) > 0 {
				err = streamHubService.RemoveStreamHub(&StreamHubDto.StreamHubReq{TenantId: tenantId, Id: channelTask.TaskId})
				if err != nil {
					e.Log.Errorf("RemoveStreamHub error:%s \r\n", err)
					return err
				}
			}

			err = e.Orm.Model(&live_schedule.CovertTask{}).Where("channel_id=? and media_id=?", channelId, deleteData.Node.Id).Delete(&channelTask).Error
			if err != nil {
				e.Log.Errorf("delete CovertTask error:%s \r\n", err)
				return err
			}

		}
	}
	data, err := json.Marshal(&newChartData)
	if err != nil {
		e.Log.Errorf("Marshal oldChartData error:%s \r\n", err)
		return err
	}
	err = e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", channelId).Update("chart_data", string(data)).Error
	if err != nil {
		e.Log.Errorf("Update ChannelBasic error:%s \r\n", err)
		return err
	}
	return nil
}

func (e *ChannelTaskService) UpdateNode(oldChartData live_schedule.ChartData, newChartData live_schedule.ChartData, basic live_schedule.ChannelBasic, newPortUrlMap map[string]string, streamHubService *streamHubService.StreamHubService, transCodeTaskService *TransCodeTaskService, convertTaskService *ConvertTaskService) error {
	e.Log.Trace("进入更新节点")
	oldPortUrlMap := make(map[string]string)
	for _, nodeData := range oldChartData.NodeData {
		nodeData := nodeData
		for _, outUrl := range nodeData.Node.Data.OutUrls {
			oldPortUrlMap[outUrl.PortId] = outUrl.OutputUrl
		}
	}

	oldNodeInUrlMap := make(map[string][]string)
	oldUrlMap := make(map[string]struct{})
	for _, oldData := range oldChartData.NodeData {
		oldData := oldData
		for _, item := range oldData.Node.Ports.Items {
			item := item
			if item.Group == "left" {
				for _, edgeData := range oldChartData.EdgeData {
					if edgeData.Target.Port == item.Id {
						url := oldPortUrlMap[edgeData.Source.Port]
						urls, ok := oldNodeInUrlMap[oldData.Node.Id]
						if !ok {
							urls = make([]string, 0, 3)
						}
						_, ok = oldUrlMap[url]
						if ok {
							continue
						}
						urls = append(urls, url)
						oldUrlMap[url] = struct{}{}
						oldNodeInUrlMap[oldData.Node.Id] = urls
					}
				}
			}
		}
	}
	newNodeInUrlMap := make(map[string][]string)
	newUrlMap := make(map[string]struct{})
	nodeMap := make(map[string]live_schedule.NodeData)
	for _, newData := range newChartData.NodeData {
		newData := newData
		nodeMap[newData.Node.Id] = newData

		for _, item := range newData.Node.Ports.Items {
			item := item
			if item.Group == "left" {
				for _, edgeData := range newChartData.EdgeData {
					if edgeData.Target.Port == item.Id {
						url := newPortUrlMap[edgeData.Source.Port]
						urls, ok := newNodeInUrlMap[newData.Node.Id]
						if !ok {
							urls = make([]string, 0, 3)
						}
						_, ok = newUrlMap[url]
						if ok {
							continue
						}
						urls = append(urls, url)
						newUrlMap[url] = struct{}{}
						newNodeInUrlMap[newData.Node.Id] = urls
					}
				}
			}
		}
	}
	e.Log.Trace("旧的数据：", oldNodeInUrlMap)
	e.Log.Trace("新的数据：", newNodeInUrlMap)

	//判断节点输入有无变更
	notEqlNodes := make([]string, 0, 3)
	notEqlNodeMap := make(map[string]struct{})
	for nodeId, newNodeInUrls := range newNodeInUrlMap {
		node := nodeMap[nodeId]
		oldNodeInUrls, ok := oldNodeInUrlMap[nodeId]
		//if !ok && node.Node.Data.Type != "media_switch" {
		//	continue
		//}

		if !ok && node.Node.Data.Type != "media_switch" {
			continue
		}

		for _, newUrl := range newNodeInUrls {
			isEql := false
			for _, oldUrl := range oldNodeInUrls {
				if newUrl == oldUrl {
					isEql = true
				}
			}
			_, ok = notEqlNodeMap[nodeId]
			if ok {

				continue
			}
			if !isEql {
				notEqlNodeMap[nodeId] = struct{}{}
				notEqlNodes = append(notEqlNodes, nodeId)
			}
		}
		_, ok = notEqlNodeMap[nodeId]
		if ok {
			e.Log.Trace("2222222222222:", ok, notEqlNodeMap)
			continue
		}

		if len(newNodeInUrls) != len(oldNodeInUrls) {
			notEqlNodeMap[nodeId] = struct{}{}
			notEqlNodes = append(notEqlNodes, nodeId)
		}
	}

	e.Log.Trace("需要更新的节点：", notEqlNodes)
	for _, nodeId := range notEqlNodes {
		nodeId := nodeId
		for _, newNode := range newChartData.NodeData {
			if newNode.Node.Id != nodeId {
				continue
			}
			inputModels := make([]live_schedule.InputModel, 0, 5)
			for _, edgeData := range newChartData.EdgeData {
				if edgeData.Target.Id == nodeId {
					inputModel := live_schedule.InputModel{}
					inputModel.InputId = edgeData.Source.Id
					inputModel.InputPort = edgeData.Source.Port
					inputModel.SelfPort = edgeData.Target.Port
					inputModel.SelfId = edgeData.Target.Id
					inputModel.InputUrl = newPortUrlMap[edgeData.Source.Port]
					inputModels = append(inputModels, inputModel)
				}
			}

			switch newNode.Node.Data.Type {
			case "media_switch":
				switchTask := live_schedule.SwitchTask{}
				err := e.Orm.Model(&live_schedule.SwitchTask{}).Where("channel_id=? and media_id=?", basic.Id, nodeId).Find(&switchTask).Error
				if err != nil {
					e.Log.Errorf("get SwitchTask error:%s \r\n", err)
					return err
				}
				if switchTask.Id > 0 {
					_, err = e.MediaSwitch(&switchTask, inputModels, newNode.Node, streamHubService, basic)
					if err != nil {
						e.Log.Errorf("MediaSwitch error:%s \r\n", err)
						return err
					}
				}

			case "media_trans":
				channelTransTask := live_schedule.ChannelTransTask{}
				err := e.Orm.Model(&live_schedule.ChannelTransTask{}).Where("channel_id=? and media_id=?", basic.Id, newNode.Node.Id).Find(&channelTransTask).Error
				if err != nil {
					e.Log.Errorf("get ChannelTransTask error:%s \r\n", err)
					return err
				}
				if channelTransTask.Id > 0 {
					_, err = transCodeTaskService.CreateTransCodeTask(newNode.Node, &channelTransTask, inputModels, basic)
					if err != nil {
						e.Log.Errorf("CreateTransCodeTask error:%s \r\n", err)
						return err
					}
				}

			case "output":
				covertTask := live_schedule.CovertTask{}
				err := e.Orm.Model(&live_schedule.CovertTask{}).Where("channel_id=? and media_id=?", basic.Id, newNode.Node.Id).Find(&covertTask).Error
				if err != nil {
					e.Log.Errorf("get ChannelTask error:%s \r\n", err)
					return err
				}
				if covertTask.Id > 0 {
					_, err = convertTaskService.CreateChannelTask1(newNode.Node, &covertTask, inputModels, basic)
					if err != nil {
						e.Log.Errorf("CreateChannelTask1 error:%s \r\n", err)
						return err
					}
				}
			default:

			}

		}
	}
	return nil
}

func (e *ChannelTaskService) StartOrStopNode(c *dto.StartOrStopNodeReq) error {
	streamHubService := &streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	streamHubService.Orm = e.Orm
	transCodeTaskService := &TransCodeTaskService{}
	transCodeTaskService.Log = e.Log
	transCodeTaskService.Orm = e.Orm
	basic := live_schedule.ChannelBasic{}
	sourceBasicService := SourceBasicService{}
	sourceBasicService.Log = e.Log
	sourceBasicService.Orm = e.Orm
	channelStatisticsService := &ChannelStatisticsService{}
	channelStatisticsService.Log = e.Log
	channelStatisticsService.Orm = e.Orm
	err := e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id = ?", c.ChannelId).Find(&basic).Error
	if err != nil {
		e.Log.Errorf("get ChannelBasic error:%s \r\n", err)
		return err
	}

	chartData := &live_schedule.ChartData{}
	if len(basic.ChartData) > 0 {
		err = json.Unmarshal([]byte(basic.ChartData), &chartData)
		if err != nil {
			e.Log.Errorf("Unmarshal ChartData error:%s \r\n", err)
			return err
		}
	}

	state := 0
	if c.Type == 1 {
		state = 1
	} else {
		state = 2
	}
	now := time.Now().Unix()
	for k, nodeData := range chartData.NodeData {
		if nodeData.Node.Id == c.NodeId {
			switch nodeData.Node.Data.Type {
			case "input":
				sourceBasic := live_schedule.SourceBasic{}
				err = e.Orm.Model(&live_schedule.SourceBasic{}).Where("id=?", nodeData.Node.Data.FormData.SourceId).Find(&sourceBasic).Error
				if err != nil {
					e.Log.Errorf("get SourceBasic error:%s \r\n", err)
					return err
				}

				if len(sourceBasic.TaskId) > 0 {
					if c.Type == 1 {
						err = sourceBasicService.StartSource(&dto.StartSourceReq{Id: sourceBasic.Id})
						if err != nil {
							e.Log.Errorf("StartSource error:%s \r\n", err)
							return err
						}
					} else {
						err = sourceBasicService.StopSource(&dto.StopSourceReq{Id: sourceBasic.Id})
						if err != nil {
							e.Log.Errorf("StopSource error:%s \r\n", err)
							return err
						}
					}
				}

			case "media_switch":
				switchTask := live_schedule.SwitchTask{}
				err = e.Orm.Model(&live_schedule.SwitchTask{}).Where("channel_id=? and media_id=?", basic.Id, c.NodeId).Find(&switchTask).Error
				if err != nil {
					e.Log.Errorf("get SwitchTask error:%s \r\n", err)
					return err
				}
				if switchTask.Id > 0 {
					if c.Type == 1 {
						err = streamHubService.StartSwitchSource(&StreamHubDto.StartOrStopSwitchSourceReq{TenantId: basic.TenantId, Id: switchTask.TaskId})
						if err != nil {
							e.Log.Errorf("StartSwitchSource error:%s \r\n", err)
							return err
						}
					} else {
						err = streamHubService.StopSwitchSource(&StreamHubDto.StartOrStopSwitchSourceReq{TenantId: basic.TenantId, Id: switchTask.TaskId})
						if err != nil {
							e.Log.Errorf("StopSwitchSource error:%s \r\n", err)
							return err
						}
					}
				}
				if c.Type == 2 {
					err = e.Orm.Model(&live_schedule.SwitchTask{}).Where("channel_id=? and media_id=?", basic.Id, c.NodeId).Update("task_state", 3).Error
					if err != nil {
						e.Log.Errorf("Update SwitchTask error:%s \r\n", err)
						return err
					}
				} else {
					err = e.Orm.Model(&live_schedule.SwitchTask{}).Where("channel_id=? and media_id=?", basic.Id, c.NodeId).Update("task_state", 1).Error
					if err != nil {
						e.Log.Errorf("Update SwitchTask error:%s \r\n", err)
						return err
					}
				}

			case "media_trans":
				channelTransTask := live_schedule.ChannelTransTask{}
				err = e.Orm.Model(&live_schedule.ChannelTransTask{}).Where("channel_id=? and media_id=?", basic.Id, c.NodeId).Find(&channelTransTask).Error
				if err != nil {
					e.Log.Errorf("get ChannelTransTask error:%s \r\n", err)
					return err
				}
				if len(channelTransTask.TaskId) > 0 {
					if c.Type == 1 {
						err = streamHubService.StartTransCode(&StreamHubDto.TransCodeReq{TenantId: basic.TenantId, Id: channelTransTask.TaskId})
						if err != nil {
							e.Log.Errorf("StartTransCode error:%s \r\n", err)
							return err
						}
						//创建调度统计记录
						err = channelStatisticsService.CreateScheduleRecord(channelTransTask.TenantId, channelTransTask.ChannelId, channelTransTask.TaskId, 2, now)
						if err != nil {
							e.Log.Errorf("CreateScheduleRecord error:%s \r\n", err)
							return err
						}
					} else {
						err = streamHubService.StopTransCode(&StreamHubDto.TransCodeReq{TenantId: basic.TenantId, Id: channelTransTask.TaskId})
						if err != nil {
							e.Log.Errorf("StopTransCode error:%s \r\n", err)
							return err
						}
						//更新任务记录结束时间
						param2 := make(map[string]interface{})
						param2["end_time"] = now
						err = e.Orm.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id = ? and end_time = 0 and type =2", channelTransTask.TaskId).Updates(&param2).Error
						if err != nil {
							e.Log.Errorf("Updates ScheduleRecord error:%s \r\n", err)
							return err
						}
					}
				}
				if c.Type == 2 {
					err = e.Orm.Model(&live_schedule.ChannelTransTask{}).Where("channel_id=? and media_id=?", basic.Id, c.NodeId).Update("task_state", 3).Error
					if err != nil {
						e.Log.Errorf("Update ChannelTransTask error:%s \r\n", err)
						return err
					}
				} else {
					err = e.Orm.Model(&live_schedule.ChannelTransTask{}).Where("channel_id=? and media_id=?", basic.Id, c.NodeId).Update("task_state", 1).Error
					if err != nil {
						e.Log.Errorf("Update ChannelTransTask error:%s \r\n", err)
						return err
					}
				}
			case "output":
				channelTask := live_schedule.CovertTask{}
				err = e.Orm.Model(&live_schedule.CovertTask{}).Where("channel_id=? and media_id=?", basic.Id, c.NodeId).Find(&channelTask).Error
				if err != nil {
					e.Log.Errorf("get CovertTask error:%s \r\n", err)
					return err
				}
				if len(channelTask.TaskId) > 0 {
					if c.Type == 1 {
						err = streamHubService.StartStreamHub(&StreamHubDto.StreamHubReq{TenantId: basic.TenantId, Id: channelTask.TaskId})
						if err != nil {
							e.Log.Errorf("StartStreamHub error:%s \r\n", err)
							return err
						}
						//创建调度统计记录
						err = channelStatisticsService.CreateScheduleRecord(channelTask.TenantId, channelTask.ChannelId, channelTask.TaskId, 1, now)
						if err != nil {
							e.Log.Errorf("CreateScheduleRecord error:%s \r\n", err)
							return err
						}
					} else {
						err = streamHubService.StopStreamHub(&StreamHubDto.StreamHubReq{TenantId: basic.TenantId, Id: channelTask.TaskId})
						if err != nil {
							e.Log.Errorf("StopStreamHub error:%s \r\n", err)
							return err
						}
						//更新任务记录结束时间
						param2 := make(map[string]interface{})
						param2["end_time"] = now
						err = e.Orm.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id = ? and end_time = 0 and type =1", channelTask.TaskId).Updates(&param2).Error
						if err != nil {
							e.Log.Errorf("Updates ScheduleRecord error:%s \r\n", err)
							return err
						}
					}
				}
				if c.Type == 2 {
					err = e.Orm.Model(&live_schedule.CovertTask{}).Where("channel_id=? and media_id=?", basic.Id, c.NodeId).Update("task_state", 3).Error
					if err != nil {
						e.Log.Errorf("get SourceBasic error:%s \r\n", err)
						return err
					}
				}

			default:

			}
			*(&chartData.NodeData[k].Node.Data.NodeState) = state
		}
	}
	if len(chartData.NodeData) > 0 {
		chartDataStr, err := json.Marshal(&chartData)
		if err != nil {
			e.Log.Errorf("Marshal ChartData error:%s \r\n", err)
			return err
		}
		err = e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", c.ChannelId).Update("chart_data", chartDataStr).Error
		if err != nil {
			e.Log.Errorf("Update ChannelBasic error:%s \r\n", err)
			return err
		}
	}

	return nil

}

func (e *ChannelTaskService) ProcessBlackWhite1(taskIds []string, tenantId string, node live_schedule.Node) error {

	streamHubService := &streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	streamHubService.Orm = e.Orm
	params := make([]StreamHubDto.SetBlackWhiteParam, 0, 5)

	for _, v := range taskIds {
		v := v
		param := StreamHubDto.SetBlackWhiteParam{}
		param.TenantId = tenantId
		param.TaskId = v
		if node.Data.Protocol != "hls" {
			continue
		}
		if node.Data.EnableBlackWhite == 2 {
			param.Acl = append(param.Acl, "allow all")
		} else if node.Data.ListType == 1 {
			for _, ip := range node.Data.IPAddresses {
				param.Acl = append(param.Acl, "deny "+ip)
			}
		} else {
			for _, ip := range node.Data.IPAddresses {
				param.Acl = append(param.Acl, "allow "+ip)
			}
		}
		params = append(params, param)
	}

	for _, v := range params {
		v := v
		if len(v.TaskId) > 0 {
			err := streamHubService.SetHlsWhiteBlack(&v)
			if err != nil {
				e.Log.Errorf("SetHlsWhiteBlack error:%s \r\n", err)
			}
		}
	}
	return nil
}

func (e *ChannelTaskService) SwitchSource(c *dto.SwitchSourceReq) error {
	switchTask := &live_schedule.SwitchTask{}
	err := e.Orm.Model(&live_schedule.SwitchTask{}).Where("channel_id=? and media_id=?", c.ChannelId, c.NodeId).Find(&switchTask).Error
	if err != nil {
		e.Log.Errorf("get SwitchTask error:%s \r\n", err)
		return err
	}

	channelBasic := live_schedule.ChannelBasic{}
	err = e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", c.ChannelId).Find(&channelBasic).Error
	if err != nil {
		e.Log.Errorf("get ChannelBasic error:%s \r\n", err)
		return err
	}
	chartData := live_schedule.ChartData{}
	if len(channelBasic.ChartData) > 0 {
		err = json.Unmarshal([]byte(channelBasic.ChartData), &chartData)
		if err != nil {
			e.Log.Errorf("Unmarshal ChartData error:%s \r\n", err)
			return err
		}
		for k, datum := range chartData.NodeData {
			if datum.Node.Id == c.NodeId {
				*(&chartData.NodeData[k].Node.Data.ActiveIndex) = c.Active
			}

		}
		chartDataStr, err := json.Marshal(&chartData)
		if err != nil {
			e.Log.Errorf("Marshal ChartData error:%s \r\n", err)
			return err
		}
		err = e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", c.ChannelId).Update("chart_data", chartDataStr).Error
		if err != nil {
			e.Log.Errorf("Update ChannelBasic error:%s \r\n", err)
			return err
		}
	}

	streamHubService := &streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	streamHubService.Orm = e.Orm
	if len(switchTask.TaskId) > 0 {
		err = streamHubService.SwitchSource(&StreamHubDto.ChangeSourceReq{Active: c.Active, TenantId: switchTask.TenantId, Id: switchTask.TaskId})
		if err != nil {
			e.Log.Errorf("SwitchSource error:%s \r\n", err)
			return err
		}
	}
	return nil

}

func (e *ChannelTaskService) SaveTrafficOff1(c *dto.TrafficOffSaveReq) error {
	tx := e.Orm.Begin()
	defer func() {
		if r := recover(); r != nil {
			e.Log.Errorf("Service SaveTrafficOff error:%s \r\n", r)
			tx.Rollback()
			//return
		}
	}()
	if err := tx.Error; err != nil {
		return err
	}
	var err error

	isDiff := false
	streamHubService := streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	streamHubService.Orm = e.Orm
	convertTaskService := ConvertTaskService{}
	convertTaskService.Log = e.Log
	convertTaskService.Orm = e.Orm

	//获取通道基本信息
	var channelBasic live_schedule.ChannelBasic
	err = tx.Model(live_schedule.ChannelBasic{}).Where("id = ?", c.ChannelId).Find(&channelBasic).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("get ChannelBasic error:%s \r\n", err)
		return err
	}
	chartData := live_schedule.ChartData{}
	if len(channelBasic.ChartData) > 0 {
		err = json.Unmarshal([]byte(channelBasic.ChartData), &chartData)
		if err != nil {
			tx.Rollback()
			e.Log.Errorf("Marshal ChartData error:%s \r\n", err)
			return err
		}
	}

	nodeInfo := live_schedule.NodeData{}
	n := 0
	for k, datum := range chartData.NodeData {
		datum := datum
		k := k
		if datum.Node.Id == c.NodeId {
			nodeInfo = datum
			n = k
			break
		}
	}
	//if len(channelBasic.NewChartData) > 0 {
	//	err = json.Unmarshal([]byte(channelBasic.NewChartData), &c.ChartData)
	//	if err != nil {
	//		tx.Rollback()
	//		e.Log.Errorf("Unmarshal ChartData error:%s \r\n", err)
	//		return err
	//	}
	//}

	//比较黑白名单是否变化
	originIps := make([]string, 0, 3)
	err = tx.Model(&live_schedule.HlsBlackWhite{}).Select("ip_address").Where("channel_id=? and node_id=?", chartData.Id, c.NodeId).Find(&originIps).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("get HlsBlackWhite error:%s \r\n", err)
		return err
	}
	if nodeInfo.Node.Data.EnableBlackWhite != c.EnableBlackWhite || nodeInfo.Node.Data.ListType != c.ListType || len(originIps) != len(c.IPAddresses) {
		nodeInfo.Node.Data.EnableBlackWhite = c.EnableBlackWhite
		nodeInfo.Node.Data.ListType = c.ListType
		nodeInfo.Node.Data.IPAddresses = c.IPAddresses
		isDiff = true
	}
	for _, v := range c.IPAddresses {
		isSame := false
		for _, ip := range originIps {
			if ip == v {
				isSame = true
			}
		}
		if !isSame {
			isDiff = true
			break
		}
	}

	taskIds := make([]string, 0, 5)
	//处理被白名单
	err = tx.Model(&live_schedule.HlsBlackWhite{}).Where("channel_id=? and node_id=?", chartData.Id, c.NodeId).Delete(&live_schedule.HlsBlackWhite{}).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("delete HlsBlackWhite error:%s \r\n", err)
		return err
	}
	hlsBlackWhites := make([]live_schedule.HlsBlackWhite, 0, 5)
	for _, v := range c.IPAddresses {
		v := v
		hlsBlackWhite := live_schedule.HlsBlackWhite{}
		hlsBlackWhite.TenantId = c.TenantId
		hlsBlackWhite.ChannelId = c.ChannelId
		hlsBlackWhite.OffId = 0
		hlsBlackWhite.NodeId = c.NodeId
		hlsBlackWhite.IpAddress = v
		hlsBlackWhites = append(hlsBlackWhites, hlsBlackWhite)
	}
	if len(hlsBlackWhites) > 0 {
		err = tx.Model(&live_schedule.HlsBlackWhite{}).Create(&hlsBlackWhites).Error
		if err != nil {
			tx.Rollback()
			e.Log.Errorf("create HlsBlackWhite error:%s \r\n", err)
			return err
		}
	}

	//now := time.Now().Unix()
	//scheduleIds := make([]string, 0, 5)

	convertTask := live_schedule.CovertTask{}
	err = e.Orm.Model(&live_schedule.CovertTask{}).Where("channel_id=? and media_id=?", c.ChannelId, c.NodeId).Find(&convertTask).Error
	if err != nil {
		e.Log.Errorf("get CortexTask error:%s \r\n", err)
		return err
	}

	if (len(c.PushInfos) > 0 || len(c.PullInfos) > 0) && convertTask.Id > 0 {

		taskIds = append(taskIds, convertTask.OutNodeId)
		//如果存在，并且没有转推任务直接保存
		if len(c.PushInfos) > 0 && len(convertTask.TaskId) <= 0 {
			*(&convertTask.PushUrl) = c.PushInfos[0].PushUrl
			nodeInfo.Node.Data.SourceUrl = c.PushInfos[0].PushUrl
		}

		//如果存在，并且有转推任务 需要先清除任务
		if len(c.PushInfos) > 0 && len(convertTask.TaskId) > 0 && c.PushInfos[0].PushUrl != convertTask.OutUrl {
			err = streamHubService.StopStreamHub(&StreamHubDto.StreamHubReq{TenantId: convertTask.TenantId, Id: convertTask.TaskId})
			if err != nil {
				tx.Rollback()
				e.Log.Errorf("StopStreamHub error:%s \r\n", err)
				return err
			}
			err = streamHubService.DeleteStreamHub(&StreamHubDto.StreamHubReq{TenantId: convertTask.TenantId, Id: convertTask.TaskId})
			if err != nil {
				tx.Rollback()
				e.Log.Errorf("DeleteStreamHub error:%s \r\n", err)
				return err
			}
			//scheduleIds = append(scheduleIds, v.TaskId)
			*(&convertTask.TaskId) = ""
			*(&convertTask.PushUrl) = c.PushInfos[0].PushUrl
			nodeInfo.Node.Data.SourceUrl = c.PushInfos[0].PushUrl
			inputModels := make([]live_schedule.InputModel, 0, 3)
			inputModels = append(inputModels, live_schedule.InputModel{InputUrl: convertTask.InUrl})
			convertTask, err = convertTaskService.CreateChannelTask1(nodeInfo.Node, &convertTask, inputModels, channelBasic)
			if err != nil {
				tx.Rollback()
				e.Log.Errorf("CreateChannelTask1 error:%s \r\n", err)
				return err
			}
			nodeInfo.Node.Data.TaskId = convertTask.TaskId
		}

		if len(c.PullInfos) > 0 {
			arr := strings.Split(c.PullInfos[0].PullUrl, "/")
			if len(arr) > 1 {
				index := strings.Index(arr[1], ".")
				sub := arr[1][0:index]
				//如果编辑自定义参数且任务是启用状态需要调用编辑自定义参数接口 && channelBasic.ChannelState != 2
				if len(convertTask.TaskId) > 0 {
					err = streamHubService.EditRefinePlayUrl(&StreamHubDto.EditRefinePlayUrlReq{TenantId: c.TenantId, Id: convertTask.OutNodeId, RedefinePlayUrlEnable: "1", RedefinePlayUrlParams: StreamHubDto.RedefinePlayUrlParams{Path: arr[0], Sub: sub}})
					if err != nil {
						tx.Rollback()
						e.Log.Errorf("EditRefinePlayUrl error:%s \r\n", err)
						return err
					}
				}
			}
			*(&convertTask.CustomOutUrl) = c.PullInfos[0].PullUrl
			nodeInfo.Node.Data.CustomUrl = c.PullInfos[0].PullUrl
		}
		err = tx.Save(&convertTask).Error
		if err != nil {
			tx.Rollback()
			e.Log.Errorf("save cortexTasks error:%s \r\n", err)
			return err
		}
	}

	//if len(channelTasks) > 0 {
	//	err = tx.Save(&channelTasks).Error
	//	if err != nil {
	//		tx.Rollback()
	//		e.Log.Errorf("save ChannelTask error:%s \r\n", err)
	//		return err
	//	}
	//}
	// 自定义转成非自定义
	if nodeInfo.Node.Data.IsCustom == 1 && c.IsCustom == 2 {
		nodeInfo.Node.Data.IsCustom = c.IsCustom
		err = streamHubService.EditRefinePlayUrl(&StreamHubDto.EditRefinePlayUrlReq{TenantId: c.TenantId, Id: convertTask.OutNodeId, RedefinePlayUrlEnable: "2", RedefinePlayUrlParams: StreamHubDto.RedefinePlayUrlParams{Path: "", Sub: ""}})
		if err != nil {
			tx.Rollback()
			e.Log.Errorf("EditRefinePlayUrl error:%s \r\n", err)
			return err
		}
	}
	//if len(scheduleIds) > 0 {
	//	//更新任务记录结束时间
	//	param := make(map[string]interface{})
	//	param["end_time"] = now
	//	err = tx.Model(live_schedule.ScheduleRecord{}).Where("channel_task_id in ? and end_time = 0", scheduleIds).Updates(&param).Error
	//	if err != nil {
	//		e.Log.Errorf("Updates ScheduleRecord error:%s \r\n", err)
	//		tx.Rollback()
	//		return err
	//	}
	//}

	//处理黑白名单
	if isDiff {
		covertTasks := make([]live_schedule.CovertTask, 0, 5)
		err = tx.Model(live_schedule.CovertTask{}).Where("channel_id=? and media_id=?", c.ChannelId, c.NodeId).Find(&covertTasks).Error
		if err != nil {
			tx.Rollback()
			e.Log.Errorf("get CovertTask error:%s \r\n", err)
			return err
		}
		for _, v := range covertTasks {
			v := v
			taskIds = append(taskIds, v.OutNodeId)
		}
		e.ProcessBlackWhite1(taskIds, channelBasic.TenantId, nodeInfo.Node)
	}

	nodeInfo.Node.Data.SourceName = c.TrafficOffName
	*(&chartData.NodeData[n]) = nodeInfo
	chartDataStr, _ := json.Marshal(&chartData)
	param := make(map[string]interface{})
	param["chart_data"] = chartDataStr
	err = tx.Model(&live_schedule.ChannelBasic{}).Where("id=?", c.ChannelId).Updates(&param).Error
	if err != nil {
		tx.Rollback()
		e.Log.Errorf("update ChannelBasic error:%s \r\n", err)
		return err
	}
	tx.Commit()
	return nil
}

func (e *ChannelTaskService) SaveMediaProcess1(c *dto.MediaProcessSaveReq1) error {

	transCodeTaskService := &TransCodeTaskService{}
	transCodeTaskService.Log = e.Log
	transCodeTaskService.Orm = e.Orm
	streamHubService := streamHubService.StreamHubService{}
	streamHubService.Log = e.Log
	streamHubService.Orm = e.Orm
	var channelBasic live_schedule.ChannelBasic
	err := e.Orm.Model(live_schedule.ChannelBasic{}).Where("id = ?", c.ChannelId).Find(&channelBasic).Error
	if err != nil {
		e.Log.Errorf("get ChannelBasic error:%s \r\n", err)
		return err
	}
	chartData := live_schedule.ChartData{}
	if len(channelBasic.ChartData) > 0 {
		err = json.Unmarshal([]byte(channelBasic.ChartData), &chartData)
		if err != nil {
			e.Log.Errorf("Marshal ChartData error:%s \r\n", err)
			return err
		}
	}

	nodeInfo := live_schedule.NodeData{}
	n := 0
	for k, datum := range chartData.NodeData {
		datum := datum
		k := k
		if datum.Node.Id == c.NodeData.Node.Id {
			nodeInfo = datum
			n = k
			break
		}
	}

	switch nodeInfo.Node.Data.Type {
	case "media_switch":
		nodeInfo.Node.Data.InputNum = c.NodeData.Node.Data.InputNum
		nodeInfo.Node.Ports.Items = make([]live_schedule.Items, 0, 3)
		nodeInfo.Node.Ports.Items = c.NodeData.Node.Ports.Items
		nodeInfo.Node.Ports = c.NodeData.Node.Ports
	case "media_trans":
		if nodeInfo.Node.Data.CustomUrl != c.NodeData.Node.Data.CustomUrl || nodeInfo.Node.Data.Template.TemplateId != c.NodeData.Node.Data.Template.TemplateId {
			nodeInfo.Node.Data.CustomUrl = c.NodeData.Node.Data.CustomUrl
			nodeInfo.Node.Data.Template = c.NodeData.Node.Data.Template
			basic := live_schedule.ChannelBasic{}
			err = e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", c.ChannelId).Find(&basic).Error
			if err != nil {
				e.Log.Errorf("get ChannelBasic error:%s \r\n", err)
				return err
			}
			channelTransTask := live_schedule.ChannelTransTask{}
			err = e.Orm.Model(&live_schedule.ChannelTransTask{}).Where("channel_id=? and media_id=?", c.ChannelId, nodeInfo.Node.Id).Find(&channelTransTask).Error
			if err != nil {
				e.Log.Errorf("get ChannelTransTask error:%s \r\n", err)
				return err
			}
			if channelTransTask.Id > 0 {
				if len(channelTransTask.TaskId) > 0 {
					err = streamHubService.RemoveTransCodeTask(&StreamHubDto.TransCodeReq{TenantId: c.TenantId, Id: channelTransTask.TaskId})
					if err != nil {
						e.Log.Errorf("RemoveTransCodeTask error:%s \r\n", err)
						return err
					}
				}

				err = e.Orm.Model(&live_schedule.ChannelTransTask{}).Where("channel_id=? and media_id=?", c.ChannelId, nodeInfo.Node.Id).Delete(&channelTransTask).Error
				if err != nil {
					e.Log.Errorf("delete ChannelTransTask error:%s \r\n", err)
					return err
				}
				channelTransTask.TenantId = c.TenantId
				channelTransTask.ChannelId = c.ChannelId
				channelTransTask.MediaId = nodeInfo.Node.Id
				channelTransTask.TaskId = ""
				inputModels := make([]live_schedule.InputModel, 0, 3)
				inputModels = append(inputModels, live_schedule.InputModel{InputUrl: channelTransTask.InUrl})
				transCodeTask, err := transCodeTaskService.CreateTransCodeTask(nodeInfo.Node, &channelTransTask, inputModels, basic)
				if err != nil {
					e.Log.Errorf("CreateTransCodeTask error:%s \r\n", err)
					return err
				}
				nodeInfo.Node.Data.TaskId = transCodeTask.TaskId
				nodeInfo.Node.Data.OutUrls[0].OutputUrl = transCodeTask.OutUrl

			} else {
				nodeInfo.Node.Data.OutUrls[0].OutputUrl = nodeInfo.Node.Data.CustomUrl
			}
		}
	}
	*(&chartData.NodeData[n]) = nodeInfo
	chartDataStr, _ := json.Marshal(&chartData)
	param := make(map[string]interface{})
	param["chart_data"] = chartDataStr
	err = e.Orm.Model(&live_schedule.ChannelBasic{}).Where("id=?", c.ChannelId).Updates(&param).Error
	if err != nil {
		e.Log.Errorf("update ChannelBasic error:%s \r\n", err)
		return err
	}
	return nil
}
